奇富科技：大數據任務從診斷到自愈的實踐之路

一、為什麼要做診斷引擎

毓數平臺是奇富科技公司自主研發的一站式大數據管理、開發、分析平臺，覆蓋大數據資產管理、數據開發及任務調度、自助分析及可視化、統一指標管理等多個數據生命周期流程，讓用戶使用數據的同時，挖掘數據最大的價值。而毓數平臺的大數據任務調度底層是基於Apache DolphinScheduler實現的。

整個大數據平臺有1000+機器、70P數據量，每日新增200T數據。每天在毓數工作流上運行的任務實例有13萬+，周活躍用戶400+；每天在毓數自助查詢中運行的sql有16萬+，周活躍用戶500+。運行的任務類型有Spark任務、Sqoop任務、DataX任務等10多種任務類型。

而我們的幾百位業務同學對大數據框架底層原理幾乎都不太瞭解，因此日常會需要數據平臺部門同學協助業務去分析和排查大數據任務運行問題。數據平臺部門同學每天都會花費很多時間和精力在任務人工診斷上，而相應業務同學面對異常任務也會很苦惱。每個月數據平臺部門協助用戶診斷任務問題平均耗費4人/天工時。

對於，異常任務，讓我印象很深刻的一件事情是，23年有一位業務同學在群里詢問他的sql為啥報錯。將業務的Sql簡寫後異常如下圖所示。

然後過了幾分鐘另外一個業務同學在群里回覆他說：“你多寫了一個庫名”。這個現象讓我發現，雖然我們數據平臺的同學認為這個問題很簡單，很容易排查。但是當業務的sql比較複雜並且業務對寫sql不太熟悉時，就會經常被異常困擾幾分鐘甚至更長時間，不能快速解決sql的異常。

而且sql語法異常可能大家排查和解決比較簡單，而spark sql的數據傾斜、數據膨脹、OOM等異常需要業務登錄spark ui去排查。這樣對業務簡直就太困難了，因此我們決定做一款面向自助查詢和工作流的大數據任務實時診斷引擎。

二、診斷引擎需求分析

（一)、診斷所有大數據任務

我們理想中診斷引擎必須支持毓數平臺中運行的所有大數據任務。前面我們說過，我們工作流有Spark任務、Sqoop任務、DataX任務等10多種任務類型；而自助查詢支持mysql、tidb、doris、tidb、spark等數據源查詢。因此診斷引擎必須能夠對所有大數據任務進行診斷。

(二)、實時看見診斷結果

診斷引擎必須實時產出診斷報告，我們不太能接受用戶看見了異常點擊診斷按鈕，然後旋轉幾圈才能看見診斷報告（也可能看不見報告）。也不能接受用戶需要從A平臺跳轉到B平臺去看診斷報告。

用武俠小說里的一句話，“毒蛇出沒之處，七步之內必有解藥”。必須讓用戶在看見異常1秒鐘內看見所對應的診斷報告。讓用戶使用毓數平臺的診斷體驗達到最好。

其次，通過自助查詢提交的Spark任務雖然還在運行中，沒有失敗，但是從Spark指標或者日誌中已經發現了數據傾斜、數據膨脹、OOM等。也應該在任務提交處，及時彈出診斷報告。達到任務還沒有失敗時，診斷報告已經產出。最終實現異常與診斷報告實時產出。

(三)、診斷規則易擴展

診斷引擎的診斷規則新增流程必須快速，不能通過發版的方式，例如每周發佈一次。如果診斷規則，每周發佈一次，對於我們快速幫助業務解決問題而言，太慢了。

因此，我們對於診斷規則發佈和生效要求在分鐘級。開發、測試好後的診斷規則在1分鐘內就能發佈到生產環境進行診斷。

(四)、診斷規則灰度發佈

診斷引擎的診斷規則必須支持灰度發佈，在生產環境運行一段時間後，效果評估沒有問題後，用戶才能在任務提交入口處看見診斷報告。

(五)、支持閾值實時調整

比如我們的診斷規則有Spark大任務、Spark 小文件。那麼多少Spark的task達到多少閾值才算命中Spark大任務？平均寫出HDFS多少位元組算是小文件？這些閾值可以脫離診斷規則，配置在常量表中。同樣修改這些閾值配置，支持分鐘級生效。

三、診斷引擎架構剖析

（一）、大數據任務提交入口

診斷元數據採集的入口有工作流、自助查詢，但是未來也不排除需要支持Linux客戶端、Jupyter等提交任務入口。

（二）、大數據診斷元數據類型

需要支持的大數據診斷元數據有log、metrics，未來也不排除支持Trace等數據。

（三）、診斷引擎架構

以下是我們診斷引擎架構圖，我們會實時收集Apache DolphinScheduler worker上所有任務產生的日誌到Kafka中。

我們也會和我司自研的自助查詢系統合作，自助查詢系統也會實時發送用戶提交任務的日誌到Kafka中。其次，我們重寫了Spark的metric sink，實現將運行的Spark任務metric實時發送到kafka中。

我們自研了一個基於Flink和Janino的規則引擎，通過Flink消費Kafka的數據，並對數據進行實時診斷。而Flink引擎每隔1分鐘會將存在mysql中新增的java代碼交給Janino編譯成位元組碼，從而實現了規則實時載入和實時生效。

規則引擎實時載入規則，並對kafka中的數據進行實時診斷，將診斷結果寫入mysql中。而工作流或者自助查詢系統調用介面查詢mysql中是否有診斷報告，並對用戶展示診斷報告。

（四）、診斷引擎與開源對比

以下是我司自研的大數據診斷引擎與OPPO開源的羅盤對比。我覺得我們的診斷引擎在易用性、實時性，規則灰度、任務自愈、大數據任務類型等方面是優於羅盤的。不過我們在診斷規則深度方面還不如羅盤，比如Spark CPU浪費、記憶體浪費等方面，我們還沒有開發規則。

不過後續，我們也可以實現這些資源浪費規則，目前精力主要做的是異常覆蓋率提升、還有調度自愈等。

OPPO羅盤：https://github.com/cubefs/compass

四、診斷引擎實現

一）任務元數據採集

Push：有侵入性、運維簡單、容易丟數據、適應性好。一般適用於任務類採集。

Pull：沒有侵入性、運維複雜、不容易丟數據、適應性差。一般適用於服務類採集。

在對工作流任務運行的log採集上，我們採用了push方式主動上報任務運行日誌到kafka中。

不採用pull方式的原因是，worker上運行的任務log寫到了磁碟很多個文件中，而通過filebeat等方式採集磁碟log文件很難實現。

在這裡，可以將taskAppId理解為分散式跟蹤中的TraceId，每條log都需要攜帶一條TraceId，通過TraceId和工作流或者自助查詢等關聯起來。

對自助查詢也是，自助查詢提交的sql運行的日誌也會實時push到kafka中。通過解析自助查詢發送到kafka中log攜帶的TaskAppId，就能關聯到自助查詢每個Query的唯一的ID。

而採集Spark運行的指標，因為我們要做到實時診斷，因此沒有採集Spark history的數據，而是實現一個spark metric kafka sink。這樣隨著啟動Spark任務，便會將Spark任務運行的metric實時上報到kafka中。

{
    "app_name": "SPARK",
    "sparkMetricReportVersion": "v1",
    "applicationUser": "hive",
    "dataCenter": "xx",
    "ip_address": "xx",
    "metrics": [
        {
            "metric": {
                "metricLabel": {},
                "metricName": "driver_ExecutorAllocationManager_executors_numberExecutorsDecommissionUnfinished",
                "metricValue": 0
            }
        },
        {
            "metric": {
                "metricLabel": {},
                "metricName": "driver_ExecutorAllocationManager_executors_numberExecutorsExitedUnexpectedly",
                "metricValue": 0
            }
        },
        {
            "metric": {
                "metricLabel": {},
                "metricName": "driver_ExecutorAllocationManager_executors_numberExecutorsGracefullyDecommissioned",
                "metricValue": 0
            }
        }
    ],
    "application_id": "application_1691417751496_779447",
    "event_time": 1692272861862,
    "applicationName": "liukunyuan-test",
    "sparkVersion": "3.3.2"
}

以上是Spark的metric上報的部分指標，發現數據無法滿足我們診斷需求。而我們需要的是Spark UI中展示的metric數據。

通過分析知道，既然Spark UI可以展示這些指標，那麼這些指標必然存在Spark Driver中，只要我們從Driver中拿到這些指標就能發送到Kafka中了。後來我們從Spark源碼中知道，Spark UI展示的這些metric存在於AppStatusStore數據結構中。因此我們從AppStatusStore中獲取指標，併發送到kafka中。

{
    "app_name":"SPARK_METRIC",
    "sparkMetricReportVersion":"v2",
    "applicationUser":"xx",
    "dataCenter":"xx",
    "ip_address":"xx",
    "metrics":[
        {
            "metric":{
                "shufflerecordsWritten":10353530,
                "taskKey":100,
                "localBytesRead":0,
                "attempt":0,
                "duration":3900003,
                "executorDeserializeCpuTime":5917691,
                "shufflewriteTime":0,
                "resultSize":4480,
                "host":"xx",
                "peakExecutionMemory":2228224,
                "recordsWritten":0,
                "remoteBlocksFetched":0,
                "stageId":1,
                "bytesWritten":0,
                "jvmGcTime":0,
                "remoteBytesRead":3221225479,
                "executorCpuTime":16868539,
                "executorDeserializeTime":5,
                "memoryBytesSpilled":0,
                "executorRunTime":16,
                "fetchWaitTime":0,
                "errorMessage":"",
                "recordsRead":0,
                "shuffleRecordsRead":1035353,
                "bytesRead":0,
                "remoteBytesReadToDisk":0,
                "diskBytesSpilled":0,
                "shufflebytesWritten":32212254,
                "speculative":false,
                "jobId":0,
                "launchTime":1692783512714,
                "localBlocksFetched":0,
                "resultSerializationTime":0,
                "name":"task",
                "taskId":100,
                "status":"SUCCESS"
            }
        }
    ],
    "application_id":"application_1692703487691_0205",
    "event_time":1692783655749,
    "applicationName":"xx",
    "sparkVersion":"3.3.2"
}

（二）為何選擇Janino

Janino是一個超小、超快的開源Java 編譯器。Janino不僅可以像javac一樣將一組Java源文件編譯成一組位元組碼class文件，還可以在記憶體中編譯Java表達式、代碼塊、類和.java文件，載入位元組碼並直接在JVM中執行。而我們可以通過Janino動態編譯java代碼，從而實現一個輕量級規則引擎。

參考《揭秘位元組跳動埋點數據實時動態處理引擎https://www.sohu.com/a/483087518_121124379》中：“Janino編譯出的原生class性能接近原生class，是Groovy的4倍左右”。並且Flink大量使用Janino動態生成java代碼。而我編寫了一個診斷規則測試執行1千萬次Janino和java耗時。發現的確如文章所說，Janino和Java性能差不多，因此選用Janino作為診斷規則的規則引擎。

這樣，規則代碼就可以存放到mysql中，實現診斷規則動態載入。不過，需要註意的是，診斷規則編譯這些java字元串代碼時，需要md5存map中，通過md5值判斷是否已經編譯過，防止重覆編譯，否則Flink運行一段時間，就會OOM。

（三）診斷數據模型設計

對於診斷引擎數據模型應該怎樣設計呢？我首先想到的是大數據任務診斷類似於醫院體檢。

我們想象一下現實中，我們是如何體檢的。首先，醫生會採集體檢對象的血液、身高、體重、X光檢測等。這些採集到的數據，對應診斷系統中的metric指標，例如shuffle條數、gc耗時等。醫生還會詢問我們是否熬夜，最近是不是有哪裡不舒服。我們會回答，沒有不舒服的地方或者我胃不太舒服。而“胃不太舒服”就是對應診斷系統中的log。醫生拿到指標和log後，會根據metric和log以及醫生的診斷經驗（診斷規則）生成診斷報告。而診斷報告有幾個要素：疾病名字、疾病描述、命中的metric或者log、結論、藥方。

1. 病類型

病類型表：存儲疾病類型，例如胃病；疾病描述，例如此病會影響食欲甚至危急生命。對應診斷引擎就是“語法錯誤”、“數據傾斜”等。

2. 藥方

優化建議表：存儲了這個疾病通用的優化規則，例如應該吃什麼藥。對應診斷引擎就是“如何調整spark參數”、“在寫庫名時，不小心寫成了 庫名.庫名.表名，這種情況下需要改寫為 庫名.表名”等。

3. 診斷規則

診斷規則表：存儲了診斷規則，例如如何看X光機拍攝的照片，或者分析抽血得到血液指標等。對應診斷引擎中java代碼部分，運行診斷規則會生成診斷結論。

4. 診斷結論

診斷結論表：存儲了運行診斷規則得到的結論。我得了胃病，嚴不嚴重，並能關聯到藥方。

對應診斷引擎就是運行java代碼時分析metric或者log得到疾病類型、生成結論，並關聯診斷建議（藥方）。

5. 常量表

常量表：常量表存儲了診斷規則需要用到的一些閾值。例如高血壓值是多少，低血壓值是多少？對於診斷引擎就是數據傾斜多少倍數算傾斜？數據膨脹多少倍算是膨脹？平均位元組數多少算是小文件？為什麼不把閾值放診斷規則里呢？是因為考慮到診斷規則是多變的，因此把常用閾值放入常量表中，這樣就不用修改診斷規則java代碼就可以調整診斷邏輯了。

五、大數據任務診斷應用

（一）診斷效果概覽

1、診斷引擎生產應用情況

我們共開發了55條基於log或者metric的診斷規則，每天診斷的log條數約為7500萬條，每天對log診斷次數約為36億次；每天診斷的metric條數約為440萬條，每天對metric診斷次數約為2640萬次；每天生成的5.5萬份診斷報告。

自從我們2023年10月份在工作流和自助查詢上線了診斷引擎之後，平均減少了數據平臺協助用戶排查問題0.5人/天工時，工作流異常診斷覆蓋率達到了85%，自助查詢異常診斷覆蓋率達到了88%，用戶在服務群中詢問問題的次數明顯減少。

2、工作流失敗任務診斷效果

以某日工作流診斷效果來看，工作流排除了依賴失敗，共有2400多個失敗任務。其中354個失敗任務沒有診斷報告，2062個失敗任務生成了診斷報告，失敗任務診斷報告命中率85.35%。

3、工作流成功任務診斷效果

診斷引擎是根據log和metric進行診斷，因此是沒有區分成功任務還是失敗任務。因此可以根據診斷命中情況對成功任務在未來運行情況進行預測。雖然任務運行成功了，如果任務發生了記憶體溢出、數據傾斜、數據膨脹等，可能也需要用戶去關註。

還有一種情況是，數據質量的弱校驗規則，數據質量比對沒有通過。因為是弱數據質量規則，並不會導致任務失敗，不過也需要提示出來。還有“同步0記錄”診斷規則，比如DataX將Hive表數據同步到Mysql時，如果同步了0條數據，雖然同步任務成功了，我們也會在工作流頁面提示出來。

4、自助查詢失敗任務診斷效果

自助查詢失敗的任務當日有2694個，而生成了診斷報告的失敗任務有2378個，診斷命中率88.27%。業務在自助查詢平臺中提交的sql是在做探索性分析，大部分失敗是語法錯誤，也是合理的。

（二）同步異常規則

診斷規則通過字元串匹配DataX拋出的異常日誌判斷是否命中。工作流在展示工作流實例時查詢診斷命中表，並展示診斷結果。用戶點擊診斷結果會跳轉到診斷報告。

在診斷報告中會展示病癥大類、ID（application_id）、病癥描述、病癥發生時間、命中的關鍵日誌或者指標、病癥結論以及診斷建議（藥方）。比如在這個診斷建議中，就說明“數據同步，目前只支持orc格式，請創建orc格式的hive表”，用戶就明白可以將自己的Hive表格式轉為orc格式解決這個異常。

通過這種方式，用戶在查詢工作流實例時，在看見實例失敗的1秒鐘內，就能找到診斷報告去解決這個異常問題，實現了“七步之內必有解藥”。

（三）語法錯誤規則

語法錯誤在自助查詢平臺和工作流中都是很常見的異常，特別是自助查詢平臺提交的任務語法錯誤情況會非常多。因此語法錯誤這個異常大類會對應很多診斷規則，屬於一對多的情況。

比如以Spark中查詢的欄位列數量和和插入表的列數量不匹配這個情況舉例。我們在診斷規則中匹配“requires that the data to be inserted have the same number of columns as the target table”，如果log中出現了這個字元串，將會命中我們的診斷規則。

語法錯誤更多是發生在自助查詢中，以讓我印象很深刻的sql語句中多寫了一個庫名舉例，我們現在也實現了在發生這種情況時自動彈出“毓智AI專家”按鈕。業務點擊按鈕就可以看見這個異常對應的診斷報告和結論：“sql中庫名寫多了一個，庫名叫:dp_data_db。錯誤的寫法為dp_data_db.dp_data_db.。正確的庫名寫法為:dp_data_db.”。

同樣原理，我們在診斷規則中，通過多關鍵字匹配log，並從log中正則匹配出多寫的庫名。這樣我們診斷結論能精確給出用戶多寫了哪一個庫名。

同樣，我們再以group by和select 欄位不匹配這個語法異常說明。當group by後面缺少biz_type時，spark會拋出異常。

而用戶點擊診斷按鈕，會看見我們診斷引擎給出的診斷報告：“group by語法有錯誤,group by中缺少該欄位'spark_catalog.dp_data_db.lky_test.biz_type'”。在sql語句非常複雜時，該診斷能讓用戶快速定位語法錯誤問題。

我們暫時只實現了14種“語法錯誤”診斷規則，後續還可以繼續覆蓋語法異常情況。

（四）記憶體溢出規則

Spark運行經常會遇到記憶體溢出，而我們也通過診斷規則判斷log或者metric中是否有記憶體溢出相關關鍵字判斷是否發生了記憶體溢出。

我們對“記憶體溢出”這種情況，我們會給一下spark參數調優建議，例如“降低spark.sql.adaptive.shuffle.targetPostShuffleInputSize參數值.例如在sql前面加:set spark.sql.adaptive.advisoryPartitionSizeInBytes=67108864;”，或者“增加spark.sql.shuffle.partitions參數值.例如在sql前面加:set spark.sql.shuffle.partitions=1000;”等。

用戶根據診斷建議，也多次解決了spark的記憶體溢出問題。記憶體溢出診斷規則對於調度任務自愈也是必備的前置條件。後面我們會講如何根據記憶體溢出結論對調度任務做自愈的。

（五）數據膨脹規則

在開發spark任務中，用戶也經常因為數據產生膨脹導致運行失敗問題。因此我們根據spark的metric輸入和輸出指標，判斷數據膨脹倍數，並給用戶解決數據膨脹情況的診斷報告。

通過spark metric計算數據膨脹比根據log正則匹配複雜很多，以下是我們數據膨脹診斷規則流程圖。

可以發現我們診斷數據膨脹是非常複雜的，而通過Janino實現的診斷規則相容java語法，因此可以非常方便的實現一些非常複雜的診斷規則。

（六）數據傾斜規則

同樣在開發spark任務過程中，我們也會經常遇到數據傾斜問題。我們也通過診斷規則實現了數據傾斜異常診斷並給出參數緩解該問題。

六、大數據任務自愈應用

（一）任務自愈目標

我們公司的核心大數據任務必須保證按時產出，而大數據任務高峰期又是在凌晨。因此很多用戶會在凌晨值班時被告警電話叫起來處理問題。因此，我們想針對失敗任務做自愈功能，比如數據質量校驗不通過自愈、spark oom任務凌晨自愈。舉例，在用戶感知不到情況下，DolphinScheduler的worker通過對spark oom失敗任務添加記憶體從而自愈，最終減少用戶凌晨起夜率。而白天的oom任務，我們暫時沒有開啟自愈功能。

（二）任務自愈實現流程

以spark oom自愈舉例，以下是我們調度根據診斷引擎生成的診斷報告對oom任務進行自愈的流程圖。

調度系統對oom自愈會限制時間段、是否SLA（核心）任務、上一次是否因為oom導致失敗、集群資源是否空閑、資源資源是否充足等進行判斷。從而實現了在資源可控狀態下對oom任務進行記憶體資源擴容，從而讓任務嘗試自愈。

調度系統默默做了自愈，雖然讓用戶凌晨不用處理。但是，我們還是希望提示給用戶，讓用戶白天處理，避免後續任務自愈失敗。因此我們寫了一個“嘗試自愈”的診斷規則在用戶查看工作流實例時提示給用戶，後續也會在白天直接通過告警方式提示給用戶。

從診斷報告中我們可以看出該任務在凌晨3點前命中了“記憶體溢出”診斷規則，而調度系統自動對該任務的executor記憶體擴容到6144M，從而在05：31分運行成功。

（三）任務自愈落地效果

自從我們上線了oom任務自愈功能後，在生產環境取得了比較好的效果。對SLA（核心）任務和普通任務都開啟自愈功能後，調度系統每天凌晨會進行10次左右“嘗試自愈”操作。而只統計SLA(核心)任務，調度系統每周平均進行6次“嘗試自愈”操作，減少了數倉同學67%的凌晨值班告警電話。

七、診斷引擎展望

（一）提升診斷覆蓋率

目前工作流異常診斷覆蓋率能達到85%，自助查詢異常覆蓋率能達到88%。後續可以繼續新增診斷規則提升異常覆蓋率。理論上工作流異常診斷覆蓋率可以很容易達到95%，自助查詢異常覆蓋率達到98%。

（二）診斷分級

目前異常診斷沒有分級功能，用戶在查看診斷時不知道哪些診斷是“嚴重”，需要立即解決，哪些診斷屬於“警告”，可以晚一點解決。我們可以添加診斷分級功能，甚至評分功能，讓診斷有優先順序能力。

（三）一鍵參數優化

目前對於數據膨脹、數據傾斜、記憶體異常，我們有spark參數推薦。不過還需要用戶手動修改工作流進行設置，後面我們想用戶點擊“一鍵優化”能夠自動將優化參數應用到任務中。

本文由 白鯨開源 提供發佈支持！