餓了么大數據計算引擎實踐與應用

知識 06-26

餓了么BDI-大數據平台研發團隊目前共有20人左右，主要負責離線&實時Infra和平台工具開發。其中6人的離線團隊需要維護大數據集群規模如下：

Hadoop集群規模1300+

HDFS存量數據40+PB，Read 3.5 PB+/天，Write 500TB+/天

14W MR Job/天，10W Spark Job/天，25W Presto/天

此外還需要維護Hadoop、Spark、Hive、Presto等組件餓了么內部版本，解決公司400+大數據集群用戶每天面臨的各種問題。

本文主要介紹餓了么大數據團隊如何通過對計算引擎入口的統一，降低用戶接入門檻。如何讓用戶自助分析任務異常及失敗原因，以及如何從集群產生的任務數據本身監控集群計算/存儲資源消耗，監控集群狀況，監控異常任務等。

引擎入口統一

_____

目前在餓了么對外提供的查詢引擎主要有Presto、Hive和Spark，其中Spark又有SparkThrift Server和Spark SQL兩種模式，並且Kylin也在穩步試用中，Druid也正在調研中。各種計算引擎都有自身的優缺點，適用的計算場景各不相同。

從用戶角度來說，普通用戶對此沒有較強的辨識能力，學習成本會比較高。並且當用戶可以自主選擇引擎執行任務時，會優先選擇所謂的最快引擎，而這勢必會造成引擎阻塞，或者將完全不適合的任務提交到某引擎，從而降低任務成功率。

從管理角度來說，大數據集群的入口太多，將難以實現統一管理，難以實現負載均衡、許可權控制，難以掌控集群整體對外服務能力。並且當有新的計算需求需要接入，我們還需要為其部署對應的客戶端環境。

用戶使用多種計算引擎

1功能模塊

針對這種情況，餓了么大數據團隊開發了Dispatcher，該組件的主要功能如下圖所示：

Dispatcher功能模塊

用戶所有任務全部通過Dispatcher提交，在Dispatcher中我們可以做到統一的鑒權，統一的任務執行情況跟蹤。還可以做到執行引擎的自動路由，各執行引擎負載控制，以及通過引擎降級提高任務運行成功率。

2邏輯架構

Dispatcher的邏輯架構如下圖所示：

Dispatcher系統邏輯架構

目前用戶可以通過JDBC模式調用Dispatcher服務，或者直接以Driver模式運行Dispatcher。Dispatcher接收到查詢請求後，將會統一進行鑒權、引擎路由等操作將查詢提交到對應引擎。另外，Dispatcher還有SQL轉換模塊，當發生從Presto引擎降級到Spark/Hive引擎時，將會通過該模塊自動將Presto SQL轉換成HiveQL。

通過Dispatcher對查詢入口的統一，帶來的好處如下：

用戶接入門檻低，無需再去學習各引擎使用方法和優缺點，無需手動選擇執行引擎；

部署成本低，客戶端可通過JDBC方式快速接入；

統一的鑒權和監控；

降級模塊提高任務成功率；

各引擎負載均衡；

引擎可擴展。

引擎可擴展主要是指當後續接入Kylin、Druid或者其他更多查詢引擎時，可以做到用戶無感知。由於收集到了提交到集群的所有查詢，針對每一個已有查詢計劃，我們可以獲得熱度數據，知道在全部查詢中哪些表被使用次數最多，哪些表經常被關聯查詢，哪些欄位經常被聚合查詢等，當後續接入Kylin時，可以通過這些數據快速建立或優化Cube。

3SQL畫像

在Dispatcher中最核心的是SQL畫像模塊，基本流程如下圖：

SQL路由模塊

查詢提交後，通過連接HiveServer對查詢計划進行解析，可以獲取當前查詢的所有元數據信息，比如：

讀入數據量

讀入表/分區數

各類Join次數

關聯欄位多少

聚合複雜度

過濾條件

……

上述元數據信息基本上可以對每一個查詢進行精準的描述，每一個查詢可以通過這些維度的統計信息調度到不同引擎中。

Hive對SQL進行解析並進行邏輯執行計劃優化後，將會得到優化後的Operator Tree，通過explain命令可以查看。SQL畫像數據可以從這個結果收集各種不同類型的Operator操作，如下圖所示：

SQL解析示例

從直觀的理解上我們知道，讀入數據量對於引擎的選擇是很重要的。比如當讀入少量數據時，Presto執行性能最好，讀入大量數據時Hive最穩定，而當讀入中等數據量時，可以由Spark來執行。

各類計算引擎數據量-執行時間分布

在初始階段，還可以通過讀入數據量，結合Join複雜度，聚合複雜度等因素在各種計算引擎上進行測試，採用基於規則的辦法進行路由。執行過程中記錄好每一次查詢的SQL畫像數據，執行引擎，降級鏈路等數據。基於這些畫像數據，後續可以採用比如決策樹，Logistic回歸，SVM等分類演算法實現引擎的智能路由，目前餓了么大數據團隊已經開始了這方面的嘗試。

目前在餓了么的應用中，由Dispatcher統一調度的Ad Hoc查詢，由於增加了預檢查環節，以及失敗降級環節，每天總體成功率為99.95%以上，整體PT90值為300秒左右。目前Presto承擔了Ad Hoc查詢的50%流量，SparkServer模式承擔了40%流量。

充分利用集群本身數據

_____

餓了么大數據集群每天運行的Spark&MR任務25W+，這些數據詳細記錄了每一個Mapper/Reducer或者Spark的Task的運行情況，如果能夠充分利用，將會產生巨大的價值。充分利用集群本身數據，數據驅動集群建設。這些數據不僅可以有助於集群管理人員監控集群本身的計算資源、存儲資源消耗，任務性能分析，主機運行狀態。還可以幫助用戶自助分析任務運行失敗原因，任務運行性能分析等。

餓了么大數據團隊開發的Grace項目就是在這方面的一個示例。

1Grace使用場景

對集群任務運行狀況詳細數據沒有明確認識的話，很容易當出現問題時陷入困境，從監控看到集群異常後將無法繼續進一步快速定位問題。

當經常有用戶找你說，我的任務為什麼跑失敗了？我的任務為什麼跑的這麼慢？我的任務能調一下優先順序么？不要跟我說看日誌，我看不懂。我想大家內心都是崩潰的。

當監控發出NameNode異常抖動，網路飈高，block創建增加，block創建延時增大等告警時，應該如何快速定位集群運行的異常任務？

當監控發出集群中Pending的任務太多時，用戶反饋任務大面積延遲時，如何快速找到問題根本原因？

當用戶申請計算資源時，到底應該給他們分配多少資源？當用戶申請提高任務優先順序時如何用數據說話，明確優先順序到底應該調到多少？當用戶只管上線不管下線任務時，我們如何定位哪些任務是不再需要的？

還有，如何通過實時展示各BU計算資源消耗，指定BU中各用戶計算資源消耗，佔BU資源比例。以及如何從歷史數據中分析各BU任務數，資源使用比例，BU內部各用戶的資源消耗，各任務的資源消耗等。

以下示例展示一些Grace產出數據圖表。有關BU、用戶、任務級別的數據不方便展示。

1）監控隊列

從下圖可以方便的看到各隊列最大最小資源，當前已用資源，當前運行任務數，Pending任務數，以及資源使用比例等，還可以看到這些數據的歷史趨勢。

各隊列任務情況

隊列資源使用趨勢

2）任務監控

可以查看指定隊列中運行中任務的任務類型，開始時間，運行時長，消耗當前隊列資源比例，以及消耗當前BU資源比例等。可快速定位計算資源消耗多並且運行時間長的任務，快速找到隊列阻塞原因。

指定隊列任務情況

3）監控主機失敗率

可以監控集群所有主機上的Task執行失敗率。已有監控體系會對主機的CPU，磁碟，內存，網路等硬體狀況進行監控。這些硬體故障最直觀的表現就是分配在這些有問題的主機上的任務執行緩慢或者執行失敗。運行中的任務是最靈敏的反應，一旦檢測到某主機失敗率過高，可觸發快速自動下線保障業務正常執行。後續可以結合硬體監控定位主機異常原因。

主機失敗率監控

4）任務性能分析

用戶可自助進行任務性能分析。

任務性能分析

並且可以根據異常項根據以下建議自助調整。

任務自助優化方案

5）任務失敗原因分析

對於失敗的任務，用戶也可以按照以下方法快速從調度系統查看失敗原因，以及對應的解決辦法，餓了么大數據團隊會定期收集各種典型報錯信息，更新維護自助分析知識庫。

失敗原因自助分析

除此之外，我們還可以實時監控每個任務的計算資源消耗GB Hours，總的讀入寫出數據量，Shuffle數據量等。以及運行中任務的HDFS讀寫數據量，HDFS操作數等。

當出現集群計算資源不足時，可快速定位消耗計算資源多的任務。當監控出現HDFS集群抖動，讀寫超時等異常狀況時，也可通過這些數據快速定位到異常任務。

基於這些數據還可以根據各隊列任務量，任務運行資源消耗時間段分布，合理優化各隊列資源分配比例。

根據這些任務運行狀況數據建立任務畫像，監控任務資源消耗趨勢，定位任務是否異常。再結合任務產出數據的訪問熱度，還可以反饋給調度系統動態調整任務優先順序等。

2Grace架構

上述示例中使用到的數據都是通過Grace收集的。Grace是餓了么大數據團隊開發的應用，主要用於監控分析線上MR/Spark任務運行數據，監控運行中隊列及任務明細及匯總數據。邏輯架構如下：

Grace邏輯架構

Grace是通過Spark Streaming實現的，通過消費Kafka中存儲的已完成MR任務的jhist文件或Spark任務的eventlog路徑，從HDFS對應位置獲取任務運行歷史數據，解析後得到MR/Spark任務的明細數據。再根據這些數據進行一定的聚合分析，得到任務級別，Job級別，Stage級別的匯總信息。最後通過定製化的Dr-Elephant系統對任務明細數據通過啟發式演算法進行分析，從而給用戶一些直觀化的優化提示。

對於Dr-Elephant，我們也做了定製化的變動，比如將其作為Grace體系的一個組件打包依賴。從單機部署服務的模式變成了分散式實時解析模式。將其數據源切換為Grace解析到的任務明細數據。增加每個任務的ActionId跟蹤鏈路信息，優化Spark任務解析邏輯，增加新的啟發式演算法和新的監控指標等。

總結

_____

隨著大數據生態體系越來越完善，越來越多背景不同的用戶都將加入該生態圈，我們如何降低用戶的進入門檻，方便用戶快速便捷地使用大數據資源，也是需要考慮的問題。

大數據集群中運行的絕大部分任務都是業務相關，但是隨著集群規模越來越大，任務規模越來越大，集群本身產生的數據也是不容忽視的。這部分數據才是真正反映集群使用詳細情況的，我們需要考慮如何收集使用這部分數據，從數據角度來衡量、觀察我們的集群和任務。

僅僅關注於集群整體部署、性能、穩定等方面是不夠的，如何提高用戶體驗，充分挖掘集群本身數據，用數據促進大數據集群的建設，是本次分享的主題。

Q & A

Q：能簡單介紹一下調度系統嗎？管理上萬個任務不容易。

A：調度系統說起來挺複雜的。就提幾個關鍵點吧，一個是任務之間的依賴，一個是血緣關係，一個是任務與實例，還有集群反壓、分散式調度、底層環境統一。

其中血緣關係應該是必須的，因為當你集群規模大了之後，用戶配置任務時根本無法完整地添加依賴。

通過血緣系統，對任務進行解析，當用戶配置好新任務後，自動推薦前置依賴，保證任務有序運行。

Q：如何得出集群的每天讀寫規模？Hadoop有介面？

A：集群讀寫規模是通過前面介紹的Grace收集的。因為我們會分析每個mr task或者spark task的HDFS數據讀寫量。還包括spill到磁碟的數據量、shuffle write、shuffle read數據量，以及每個任務的GBHour信息。