全球短訊！日均調度 10W+ 任務實例，DolphinScheduler 在蔚來汽車一站式數據治理開發平臺的應用改造

大家好我是張金明，在蔚來汽車擔任大數據平臺研發工程師。這次和大家分享的是 Apache DolphinScheduler 在蔚來汽車一站式數據治理開發平臺的應用和改造，接下來我將從背景、應用現狀和技術改造三個方面去分享一下。

(資料圖片)

背景

業務痛點

在蔚來汽車構建一個統一的數據中臺之前，我們面臨這樣一些業務痛點和困境：

• 數據缺乏治理，數倉不規范、不完整

  • 沒有統一的數據倉庫，無全域的數據資產視圖
  • 存在數據孤島；

• 工具散亂，用戶權限不統一、學習成本高

  • 用戶需要在多個工具之間切換，導致開發效率降低
  • 底層運維成本高；

• 數據需求響應周期長，找數難、取數難

  • 無沉淀的數據資產與中臺能力，重復處理原始數據；
  • 業務數據需求從提出到獲取結果的周期長

基于這些痛點和問題，我們構建了一個公司層面的業務中臺，內部叫做 DataSight。

我們可以看到，最底下是我們的一些基礎組件；往上一層，這些基礎組件主要是支撐了一些數據接入與開發的模塊；再向上是我們的數據治理，以及數據資產與應用層。其中，Apache DolphinScheduler 這個調度器在公司主要應用于交互的模塊，就是數據開發和數據運維兩個模塊。

數據開發中，調度任務開發主要就是用到了 Apache DolphinScheduler，通過 API 和調度器進行交互。

應用現狀

作業現狀

目前，我們的機器共有 9 臺，分別是兩臺 Master機器，是8c 和 32G；六臺 Worker 機器,16c 和 64G，以及一臺 Alert 機器，8c 和 32G。

版本是更新到了 Apache DolphinScheduler 2.0.7，后續的目標是升級到 2.0.8 版本，2.0 版本已經能夠支撐我們的業務了，整體的穩定性還是比較好的。

我們其實是從 2022 年 4 月份開始才真正地在線上運行 Apache DolphinScheduler，直到今天大概運行了一年一個月多的時間，日均的調度工作流實例大概在 4w+，日均調度任務實例大概在 10w+ 左右，主要節點是 Spark 節點、SparkSQL、prestoSQL、Python 和 Shell，其中 Spark 節點占比約 70%。

目前這些節點已經能夠支撐我們的大部分業務，后續我們可能會把 DolphinScheduler 自帶的一些節點加到我們的數據開發模塊里面來。

技術改造

為了適應我們業務的需求，我們對 Apache DolphinScheduler 進行了一些技術改造。首先是穩定性方面的工作。

穩定性

• 滾動重啟+黑名單機制+精準路由

這個改造是因為我們遇到的一些痛點，首先，大家知道，DolphinScheduler 的 Worker 重啟機制在重啟時會把所有的任務給 kill 掉，然后去Restart 這個任務，把這個 kill 的任務分發到新的 Worker 機器上。這樣會導致任務執行時間較長。這不符合我們的預期。

同時，我們也無法在特定的 Worker 上進行驗證任務。

對此，我們的解決方案就是滾動重啟，在重啟某臺機器之前先下線這臺機器，也就是加上黑名單，這樣的話，Master 機器就不會給這臺下已經下線的機器去分發 worker 任務。這臺機器會在上面的任務全部處理完畢后自動上線，也就是移出這個黑名單。接下來所有的 woker 節點都按照此種方式重啟，達到平滑重啟的目的。

這樣做的好處在于不會阻塞每個任務的執行，集群在重啟的時候穩定性能得到大幅提升。

另外，我們還做了精準路由的工作。也就是在任務名后加特定后綴，實現精準路由到某臺機器上。

如圖所示，我們在這個任務后面加一個 specific dispatch-worker02 的話，那這個任務一定會被分配到Worker02 這臺機器上去。這樣的好處在于，假設我們想要去某一個功能點，我們只需要把某一臺 Worker 機器下線重啟，需要測試的功能點按照這個方式就一定能夠打到這臺特定的機器上去，實現最小范圍的灰度，有助于提高穩定性。

• 優化存儲

在存儲方面，我們痛點也很明顯，就是 process instance和task instance 這兩張表數據量是比較大的，由于我們每天的數據量比較大，目前已經達到了千萬級別，造成 MySQL 的存儲壓力比較大。另外，部分 SQL 執行時間長，業務響應變慢；而且 DDL 時會造成鎖表，導致業務不可用。

針對這些問題，我們的解決方案包括去梳理所有的慢 SQL，然后去添加合適的索引。與此同時，還有降低查詢頻率，特別是針對依賴節點。因為我們知道依賴節點每 5 秒鐘查詢一次數據庫，所以我們根據依賴節點所在的 tasks instance ID 去做一個“打散”，偶數節點每 30 秒查詢一次，奇數節點每 30 秒查詢一次，把他們分開來降低對整個數據庫的查詢壓力。

另外，為了減輕表數據量大的問題，我們也做了一個定期刪除的策略，以及定時同步歷史數據的策略。

定時刪除就是我們利用 DolphinScheduler 自身的調度能力建立兩個工作流去刪除這兩張表，保證 process instance 這張表保留兩個月的數據，task instance 這張表保留一個月的數據。同時在刪表的時候，我們要注意在非業務高峰期時去做這個動作，每次刪表的時候，batch size 要控制好，盡量不要影響線上的任務。

定時同步歷史數據，就是我們針對 process instance 這個表，依據 schedule time 按年去分表；針對 task instance 這張表，按 first submit time 按月去分表。

• Spark 任務優化

我們提交 Spark 任務的方式是通過 Sparks Submit 去提交的，它的缺點在于提交 Spark 任務后，常駐機器，導致機器內存過大，會有機器宕機的風險，worker 的運行效率較低。

我們優化了 Spark 任務提交和運行的邏輯，就是通過 Spark Submit 提交的時候添加 spark.yarn.submit.waitAppCompletion=false這個參數，這樣任務提交完以后這個進程就消失了。考慮到要保證 worker 機器任務的線程和 Spark 和 Yarn 上的狀態一致，我們間隔一定時間查詢 Spark 任務狀態，如圖所示：

這里是一個 while true 循環，首先去判斷這個任務是否超時。如果任務已經超時就會結束這個 Spark 任務，同時會 kill 掉集群上那個真正在跑的任務。

如果任務沒有超時，我們會去獲取任務的狀態，如果任務狀態是終止狀態，就直接跳出這個循環，否則會間隔一定的時間，比如 30 秒，再繼續這個 while true 循環。這種方式讓整個 worker 機器所能承載的 Spark 任務大大增加。

易用性

接下來再看一些我們在易用性方面的改造工作吧！

• 依賴節點優化

我們的依賴節點之前的痛點在于，它的使用規則不太符合用戶的需求，比如之前是單次查詢不到上游即失敗；日志內容顯示信息不全，對用戶不友好；用戶無法自定義依賴范圍。

針對這些問題，我們做的工作包括修改了查詢邏輯為繼續等待，就是說當這個任務查詢不到上游的時候，我們會繼續等待，而不是直接失敗。同時我們會也有個極端的保證，就是這個依賴節點超過 24 小時以后就讓它自動失敗，然后給用戶發一個報警。

針對依賴節點，我們也做了強制成功這樣一個小trick，并支持用戶自定義依賴范圍。

另外，我們還優化了依賴節點的日志輸出，當用戶點擊依賴節點的日志的時候，可以比較清楚地看到依賴的上游所在的空間，這個空間內任務所對應的維護人是什么，以及工作流節點是什么和完成狀態，讓用戶可以點對點地找到上游的同學，快速解決這個依賴節點卡住的問題。

• 補數任務優化

針對補數之前的痛點，比如補數任務沒有進度提示，并行補數流程實例不嚴格按照時間順序，停止并行補數任務邏輯比較麻煩等問題，我們的解決方案包括并行任務引入線程池，也就是把任務按照時間順序一個一個拋到新建的線程池里，執行完畢以后退出這個線程池，然后再放一個新的進來，達到并行補數的狀態。同時，執行時間按遞增的順序。

當我們想停止這個補數任務的時候也比較簡單，直接把這個線程池 shutdown 就行。

• 多 SQL 執行

最后是關于多 SQL 執行方面的優化。我們之前面臨的痛點包括：

• 多 SQL 需要多節點執行浪費集群資源；
• 自定義環境變量無法實現；
• 無法跟蹤 SparkSQL 的運行日志。

我們的解決方案包括拆分這條 SQL，支持多條 SQL 同時執行。

與此同時，我們可以在 SparkSQL 任務執行之前攔截執行select engine_id() as engine_id語句。如上圖所示，對于 SQL 1 和 SQL 2，之前我們會在兩個任務里面去放著，但是現在可以在一個任務節點里面放下來，它會執行兩次。同時我們可以清晰地看到這個 SparkSQL 所在的 application ID 是什么，用戶能夠清晰地根據這個 application ID 找這個業務所在的地址，了解這個作業的進度。

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

關鍵詞：