基于 Apache Druid 的實時分析平臺在愛奇藝的實踐-魔扣目錄

最近幾年大數(shù)據(jù)技術(shù)在各行各業(yè)得到廣泛應(yīng)用，為企業(yè)的運營決策和各種業(yè)務(wù)提供支持。隨著數(shù)據(jù)的增長，業(yè)務(wù)對數(shù)據(jù)時效性的要求，給企業(yè)的大數(shù)據(jù)分析帶來了巨大挑戰(zhàn)。針對海量數(shù)據(jù)的實時分析需求，近年來市場上涌現(xiàn)出眾多OLAP分析引擎。這些OLAP引擎有各自的適用場景和優(yōu)缺點，如何選擇一款合適的引擎來更快地分析數(shù)據(jù)、更高效地挖掘數(shù)據(jù)的潛在價值？

愛奇藝大數(shù)據(jù)服務(wù)團隊評估了市面上主流的OLAP引擎，最終選擇Apache Druid時序數(shù)據(jù)庫來滿足業(yè)務(wù)的實時分析需求。本文將介紹Druid在愛奇藝的實踐情況、優(yōu)化經(jīng)驗以及平臺化建設(shè)的一些思考。

愛奇藝大數(shù)據(jù)OLAP服務(wù)

愛奇藝大數(shù)據(jù)OLAP服務(wù)在2015年前主要以離線分析為主，主要基于Hive+MySQL、HBase等。2016年起引入Kylin和Impala分別支持固定報表和Ad-hoc查詢。2018年以來引入Kudu和Druid支持實時分析需求。

在引入Druid之前，業(yè)務(wù)的一些場景無法通過離線分析滿足，如廣告主想要實時基于投放效果調(diào)整投放策略、算法工程師調(diào)整模型推到線上A/B要隔天離線報表才能看到效果。這些場景都可以歸納為對海量事件流進行實時分析，經(jīng)典的解決方案有如下幾種：

離線分析：

使用Hive、Impala或者Kylin，它們一個共同的缺點是時效性差，即只能分析一天或者一小時前的數(shù)據(jù)，Kylin還面臨維度爆炸的問題

實時分析：

用ElasticSearch或OpenTSDB，由于數(shù)據(jù)結(jié)構(gòu)本質(zhì)是行存儲，聚合分析速度都比較慢；可以通過查詢緩存、OpenTSDB預(yù)計算進行優(yōu)化，但不根本解決問題；
用流任務(wù)（Spark/Flink）實時地計算最終結(jié)果，存儲在MySQL提供進一步服務(wù)；問題是每當(dāng)需求調(diào)整，如維度變更時，則需要寫新的流任務(wù)代碼；
使用Kudu和Impala結(jié)合能夠做到實時分析。在實踐過程中發(fā)現(xiàn)，Kudu受限于內(nèi)存和單機分區(qū)數(shù)，支撐海量數(shù)據(jù)成本很大；

Lambda架構(gòu)：

無論選用哪種實時或離線方案的組合，都會采用Lambda架構(gòu)，用離線數(shù)據(jù)校準(zhǔn)實時數(shù)據(jù)。這意味著從攝入、處理、查詢都需要維護兩套架構(gòu)，新增一個維度，離線和實時均需對應(yīng)修改，維護困難

以上種種方案的不足，促使我們尋找新的解決方案，最終決定采用Druid。

Apache Druid介紹

Apache Druid是針對海量事件流進行存儲和實時多維分析的開源系統(tǒng)。它具有如下特性：

實時可見：消息攝入后分鐘級查詢可見
交互查詢：查詢延時在秒級，核心思想為內(nèi)存計算和并行計算
維度靈活：支持幾十個維度任意組合，僅在索引時指定的維度查詢可見
易于變更：需求變更后調(diào)整索引配置立馬生效；
流批一體：新版本KIS模式可實現(xiàn)Exactly Once語義

上圖為Druid架構(gòu)圖，大體分為幾個模塊：

MiddleManager ：索引節(jié)點，負責(zé)實時處理消息，將其轉(zhuǎn)成列式存儲，并通過Rollup精簡數(shù)據(jù)量；索引節(jié)點定期將內(nèi)存中數(shù)據(jù)持久化為不可修改的文件（Segment），保存至HDFS保證數(shù)據(jù)不會丟失；
Historical ：歷史節(jié)點，將Segment加載到本地，負責(zé)大部分查詢的計算；
Broker ：查詢節(jié)點，將查詢分解為實時和離線部分，轉(zhuǎn)發(fā)給索引節(jié)點和歷史節(jié)點，并匯總最終的查詢結(jié)果；
Overlord ：負責(zé)索引任務(wù)管理；
Coordinator ：負責(zé)負載均衡，確保Segment在歷史節(jié)點之間盡量均衡；

Druid在愛奇藝的實踐

Druid很好地填補了愛奇藝在實時OLAP分析領(lǐng)域的空白，隨著業(yè)務(wù)實時分析需求的增加，Druid集群和業(yè)務(wù)規(guī)模也在穩(wěn)步增長。目前集群規(guī)模在數(shù)百個節(jié)點，每天處理數(shù)千億條消息，Rollup效果在10倍以上。平均每分鐘6千條查詢，P99延時一秒內(nèi)，P90延時在200毫秒內(nèi)。在建設(shè)Druid服務(wù)過程中，我們也不斷遇到規(guī)模增長帶來的性能瓶頸和穩(wěn)定性問題。

1.Coordinator瓶頸

當(dāng)時的挑戰(zhàn)是實時索引任務(wù)經(jīng)常被阻塞。Druid的Handoff總結(jié)如下，索引節(jié)點將Segment持久化到HDFS，然后Coordinator制定調(diào)度策略，將計劃發(fā)布到ZooKeeper。歷史節(jié)點從ZooKeeper獲取計劃后異步地加載Segment。當(dāng)歷史節(jié)點加載完Segment索引節(jié)點的Handoff過程才結(jié)束。這個過程中，由于Coordinator制定計劃是單線程串行的，如果一次觸發(fā)了大量Segment加載，執(zhí)行計劃制定就會很慢，從而會阻塞Handoff過程，進而索引節(jié)點所有的Slot均會被用滿。

而以下過程均會觸發(fā)大量Segment加載，在解決Coordinator調(diào)度性能瓶頸前，很容易引發(fā)故障：

• 歷史節(jié)點因硬件故障、GC、主動運維退出

• 調(diào)整Segment副本數(shù)、保留規(guī)則

通過火焰圖對Coordinator進行Profiling最終定位了問題，如下圖所示，將最耗時部分放大出來，是負載均衡策略對每個Segment要選擇一個最佳的服務(wù)器。閱讀源碼可知其過程為，加載Segment X，需要計算它和服務(wù)器的每個Segment Y的代價Cost(X, Y)，其和為服務(wù)器和Segment X的代價。假設(shè)集群有N個Segment，M個Historical節(jié)點，則一個節(jié)點宕機，有N/M個Segment需要加載，每個Segment都和剩余的N個節(jié)點計算一次代價，調(diào)度耗時和N成平方關(guān)系。

一個節(jié)點宕機調(diào)度耗時 = (N/M)個Segment * 每個Segment調(diào)度耗時 = (N/M) * N = O(N^2)

分析清楚原因后，很容易了解到Druid新很容易了解到Druid新版本提供了新的負載均衡策略(
druid.coordinator.balancer.strategy =
CachingCostBalancerStrategy )，應(yīng)用后調(diào)度性能提升了10000倍，原先一個歷史節(jié)點宕機會阻塞Coordinator1小時到2小時，現(xiàn)在30秒內(nèi)即可完成。

2.Overlord瓶頸

Overlord性能慢，我們發(fā)現(xiàn)升級到0.14后Overlord API性能較差，導(dǎo)致的后果是索引任務(wù)概率性因調(diào)用API超時而失敗。通過Jstack分析，看到大部分的HTTP線程均為阻塞態(tài)，結(jié)合代碼分析，定位到API慢的原因，如左圖所示，Tranquility會定期調(diào)用Overlord API，獲取所有RunningTasks，Overlord內(nèi)部維護了和MySQL的連接池，該連接池默認(rèn)值為8，該默認(rèn)值值過小，阻塞了API處理。解決方法是增大dbcp連接池大小。
druid.metadata.storage.connector.dbcp.maxTotal = 64

調(diào)整后，Overlord性能得到了大幅提升，Overlord頁面打開從幾十秒降低到了幾秒。但意料之外的事情發(fā)生了，API處理能力增加帶來了CPU的飆升，如右圖所示，并且隨著Tranquility任務(wù)增加CPU逐漸打滿，Overlord頁面性能又逐步降低。通過火焰圖Profile可知，CPU主要花費在getRunningTasks的處理過程，進一步分析Tranquility源碼后得知，Tranquility有一個配置項(
druidBeam.overlordPollPeriod）可以控制Tranquility輪詢該API的間隔，增大該間隔后問題得到了暫時緩解，但根本的解決方案還是將任務(wù)切換為KIS模式。

3.索引成本

Druid索引成本過高。基于Druid官方文檔，一個Druid索引任務(wù)需要3個核，一個核用于索引消息，一個核用于處理查詢，一個核用于Handoff過程。我們采用該建議配置索引任務(wù)，壓測結(jié)果是3核配置下能夠支撐百萬/分鐘的攝入。

在最初，集群所有的索引任務(wù)都是統(tǒng)一配置，但實際使用過程中，大部分的索引任務(wù)根本達不到百萬/分鐘的消息量，造成了資源大量浪費。如下圖所示，我們按照索引任務(wù)的內(nèi)存使用量從高到低排序，9 GB為默認(rèn)配置，80%的任務(wù)利用率低于1/3，即3 GB。我們以3 GB繪制一條橫線，以內(nèi)存使用最接近的任務(wù)繪制一條豎線，定義A為實際使用的內(nèi)存，B為第二象限空白部分，C為第四象限空白部分，D為第一象限空白部分，則浪費的資源 = （B+C+D）的面積。

我們思考能否采取索引任務(wù)分級的策略，定義一種新的類型索引節(jié)點 – Tiny節(jié)點。Tiny節(jié)點配置改為1 core3GB，能夠滿足80%小任務(wù)的資源需求，而default節(jié)點繼續(xù)使用 3 core9 GB的配置，滿足20%大任務(wù)的需求，在這種新的配置下，浪費的資源 = （B + C）的面積，D這一大塊被省下來。簡單地計算可知，在不增加機器的情況下，總Slots能夠增加1倍。

默認(rèn)slot資源需求為1，Tiny為1/3，調(diào)整后單位任務(wù)需要的資源 = 0.2 * 1 + 0.8 * 1/3 = 0.5

在實際操作層面，還需解決一個問題，即如何把Datasource指定給合適的Worker節(jié)點。在Druid低版本中，需要通過配置文件將每一個Datasource和Worker節(jié)點進行關(guān)聯(lián)，假設(shè)有N個Datasource，M個Worker節(jié)點，這種配置的復(fù)雜度為 N * M，且無法較好地處理Worker節(jié)點負載均衡，Worker宕機等場景。在Druid 0.17中，引入了節(jié)點Category概念，只需將Datasource關(guān)聯(lián)特定的Category，再將Category和Worker綁定，新的配置方法有2個Category，復(fù)雜度 = 2 * N + 2 * M。

4.Tranquility vs KIS

剛使用Druid時，當(dāng)時主力模式是Tranquility。Tranquility本質(zhì)上仍然是經(jīng)典的Lambda架構(gòu)，實時數(shù)據(jù)通過Tranquility攝入，離線數(shù)據(jù)通過HDFS索引覆蓋。通過離線覆蓋的方式解決消息延遲的問題，缺點是維護兩套框架。對于節(jié)點失敗的問題，Tranquility的解決方案是鏈路冗余，即同時在兩個索引節(jié)點各起一份索引任務(wù)，任一節(jié)點失敗仍有一份能夠成功，缺點是浪費了一倍的索引資源。自0.14版本起，Druid官方建議使用KIS模式索引數(shù)據(jù)，它提供了Exactly Once語義，能夠很好地實現(xiàn)流批一體。

和Tranquility的Push模式不同，KIS采取Pull模式，索引任務(wù)從Kafka拉取消息，構(gòu)建Segment。關(guān)鍵點在于最后持久化Segment的時候，KIS任務(wù)有一個數(shù)據(jù)結(jié)構(gòu)記錄了上一次持久化的Offset位置，如圖例左下角所示，記錄了每個Kafka Partition消費的Offset。在持久化時會先檢查Segment的開始Offset和元信息是否一致。如果不一致，則會放棄本次持久化，如果一致，則觸發(fā)提交邏輯。提交中，會同時記錄Segment元信息和Kafka Offset，該提交過程為原子化操作，要么都成功，要么都失敗。

KIS如何處理各個節(jié)點失敗的情況呢？假設(shè)Kafka集群失敗，由于是Pull模式，Druid在Kafka恢復(fù)后繼續(xù)從上一個Offset開始消費；假設(shè)Druid索引節(jié)點失敗，Overlord后臺的Supervisor會監(jiān)控到相應(yīng)任務(wù)狀態(tài)，在新的索引節(jié)點啟動KIS任務(wù)，由于內(nèi)存中的狀態(tài)丟失，新的KIS任務(wù)會讀取元信息，從上一次的Offset開始消費。假設(shè)是MySQL或者更新元數(shù)據(jù)過程失敗，則取決于提交的原子操作是否成功，若成功則KIS從新的Offset開始消費，失敗則從上一次Offset開始消費。

進一步看一下KIS是如何保證Exactly Once語義。其核心是保證Kafka消費的Offset連續(xù)，且每個消息都有唯一ID。Exactly Once可以分為兩個部分，一是At Least Once，由KIS檢查Offset的機制保證，一旦發(fā)現(xiàn)缺失了部分Offset，KIS會重新消費歷史數(shù)據(jù)，該過程相當(dāng)于傳統(tǒng)的離線補數(shù)據(jù)，只是現(xiàn)在由Druid自動完成了。另一個是At Most Once，只要保證Offset沒有重疊部分，則每條消息只被處理了一次。

以下是KIS在愛奇藝的一個實例，左下圖為業(yè)務(wù)消息量和昨天的對比圖，其中一個小時任務(wù)持久化到HDFS失敗了，看到監(jiān)控曲線有一個缺口。之后Druid后臺啟動了一個新的KIS任務(wù)，一段時間后，隨著KIS補錄數(shù)據(jù)完成，曲線圖恢復(fù)到右下圖所示。那么，如果業(yè)務(wù)不是一直盯著曲線看，而是定期查看的話，完全感受不到當(dāng)中發(fā)生了異常。

基于Druid的實時分析平臺建設(shè)

Druid性能很好，但在初期推廣中卻遇到很大的阻力，主要原因是Druid的易用性差，體現(xiàn)在如下幾個方面：

數(shù)據(jù)攝入需要撰寫一個索引配置，除了對數(shù)據(jù)自身的描述（時間戳、維度和度量），還需要配置Kafka信息、Druid集群信息、任務(wù)優(yōu)化信息等
查詢的時候需要撰寫一個JSON格式的查詢，語法為Druid自定義，學(xué)習(xí)成本高
返回結(jié)果為一個JSON格式的數(shù)據(jù)，用戶需自行將其處理成最終圖表、告警
報錯信息不友好，上述所有配置均通過JSON撰寫，一個簡單的逗號、格式錯誤都會引起報錯，需花費大量時間排查

為解決Druid易用性差的問題，愛奇藝自研了實時分析平臺RAP（Realtime Analysis Platform），屏蔽了Kafka、Druid、查詢的細節(jié)，業(yè)務(wù)只需描述數(shù)據(jù)格式即可攝入數(shù)據(jù)，只需描述報表樣式、告警規(guī)則，即可配置實時報表和實時告警。

RAP實時分析平臺，主要有六大特性：

全向?qū)渲茫簶I(yè)務(wù)無需手寫ETL任務(wù)
計算存儲透明：業(yè)務(wù)無需關(guān)心底層OLAP選型
豐富報表類型：支持常見的線圖、柱狀圖、餅圖等
數(shù)據(jù)延時低：從App數(shù)據(jù)采集到生成可視化報表的端到端延時在5分鐘內(nèi)，支持?jǐn)?shù)據(jù)分析師、運營等業(yè)務(wù)實時統(tǒng)計分析UV、VV、在線用戶數(shù)等
秒級查詢：大部分查詢都是秒以內(nèi)
靈活變更：更改維度后重新上線即可生效