什么是任務(wù)調(diào)度系統(tǒng)?

A job scheduler is a computer application for controlling unattended background program execution of jobs (from wikipedia)

簡單來說就是你有很多任務(wù), 彼此之間執(zhí)行必須有一定順序構(gòu)成一個DAG. 如下圖所示:

一個簡單的 DAG 示意圖

為什么需要任務(wù)調(diào)度系統(tǒng)

數(shù)據(jù)終究是要拿來計算的. 假設(shè)我們從小白開始搭建一套數(shù)據(jù)計算系統(tǒng).

• 數(shù)據(jù)計算第一階段:

  • 搭建了 Hadoop, 執(zhí)行幾個 Hive 腳本統(tǒng)計數(shù)據(jù)
  • 為了簡單, 代碼直接寫到一個 HQL 文件中, 每天晚上 12 點執(zhí)行, 計算前一天的運營數(shù)據(jù)

• 大數(shù)據(jù)系統(tǒng)建設(shè)第二階段:

  • 數(shù)據(jù)需求開始增多, 你每天加班加點的開發(fā)新的計算, 放到同一個 HQL 文件中

• 數(shù)據(jù)計算第三階段: 拆

問題:

• 有一天你發(fā)現(xiàn) HQL 文件已經(jīng)一千多行了

  • 計算邏輯復(fù)雜, 牽一發(fā)而動全身
  • 時不時會有某個計算失敗, 全部計算重新執(zhí)行很浪費時間
  • 老板看你忙不過來, 給你招了幾個小弟一起開發(fā). 需要分拆任務(wù)

• 大 HQL 文件根據(jù)邏輯拆成幾十個小的計算文件

• 每個開發(fā)負(fù)責(zé)一塊業(yè)務(wù)的計算, 使用簡單的 bash 腳本按順序調(diào)用 hive -f 執(zhí)行幾十個 HQL 文件進(jìn)行計算

• 小心安排幾十個計算的先后順序, 避免順序顛倒導(dǎo)致計算錯誤

• 計算變成了一大串鞭炮, 挨個執(zhí)行

• 數(shù)據(jù)計算第四階段: 并發(fā)

問題:

• 任務(wù)太多, 串行執(zhí)行時間太長

• 一個任務(wù)失敗, 后續(xù)任務(wù)全部等待, 效率很低

• review 所有的計算任務(wù), 找到可以并發(fā)執(zhí)行的計算任務(wù)

• 想辦法并發(fā)執(zhí)行互不相干的一些任務(wù), 計算任務(wù)僅僅依賴必須依賴的前置任務(wù), 因此計算任務(wù)由一串鞭炮 組成了一個 DAG

因此, 我們對調(diào)度系統(tǒng)的需求如下:

• 分解計算任務(wù). 計算任務(wù)不單只有 Hive
• 任務(wù)執(zhí)行之間能夠互相依賴, 前置任務(wù)失敗, 后續(xù)依賴任務(wù)不執(zhí)行
• 盡可能的并行執(zhí)行任務(wù), 縮短執(zhí)行時間
• 定時觸發(fā)計算任務(wù). daily/weekly/hourly 等
• 任務(wù)失敗報警. 對于計算失敗的重要任務(wù), 報警必不可少. 哪個數(shù)據(jù)工程師沒有半夜起來修復(fù)過失敗的計算任務(wù)?

從系統(tǒng)的角度來說, 調(diào)度系統(tǒng)應(yīng)該包含以下幾個模塊:

• Scheduler 模塊: 負(fù)責(zé)調(diào)度 DAG, 根據(jù)條件觸發(fā)任務(wù)執(zhí)行
  • 按時啟動任務(wù), 最好支持 Cron 表達(dá)式
  • 提供任務(wù)執(zhí)行結(jié)果上報接口, Executor 模塊通過該接口告知 Scheduler 任務(wù)執(zhí)行狀態(tài). 接口不一定是 restful api, 數(shù)據(jù)庫可以作為接口
  • DAG 執(zhí)行過程中, 根據(jù)已經(jīng)成功的任務(wù), 決定下一個滿足依賴條件的任務(wù)
  • 報警模塊. 任務(wù)執(zhí)行失敗后, 根據(jù)配置的規(guī)則, 給響應(yīng)的值班人員報警.
• Executor 模塊: 負(fù)責(zé)根據(jù) Scheduler 指令, 執(zhí)行對應(yīng)的任務(wù)
  • 獲取任務(wù)執(zhí)行接口. Scheduler 通過接口通知 Executor 執(zhí)行任務(wù)
  • 重試機(jī)制. 提供一定重試機(jī)制, 當(dāng)任務(wù)因為網(wǎng)絡(luò)抖動等原因失敗時自動重試, 減少報警次數(shù), 也就是減少數(shù)據(jù)工程師半夜爬起來的次數(shù) LOL.
    • 重試次數(shù)不能太多, Fail Fast 在 Batch 計算業(yè)務(wù)中也同樣重要
    • 計算任務(wù)必須具備冪等性. 例如, 一個 Hive 計算寫成如下方式

-- test.tmp_test_table 用于存儲中間計算結(jié)果
CREATE TABLE test.tmp_test_table AS
SELECT data_date,
       count(*) AS cnt
FROM test_data
GROUP BY data_date;

是否具備冪等性? 回答是否定的. 因為一旦任務(wù)執(zhí)行成功, 第二次執(zhí)行時 test.tmp_test_table 已經(jīng)存在, 任務(wù)會報錯 (Table Already Exists).
如何修改? 很簡單, 前面加一句 DROP TABLE IF EXISTS test.tmp_test_table 便可.

• 代碼部署方式. Executor 要執(zhí)行計算必須有代碼. 那如何獲取計算任務(wù)的代碼就是一個問題. 從擴(kuò)展性來說, Executor 一定要支持分布式部署, 也就是一個 Scheduler 多個Executor.
  • 一種方式是, 將計算代碼部署到每個 Executor 節(jié)點.
    采用這種方式要求使用者必須有自動化部署, 因為計算代碼漏部署或者錯誤部署了一臺 Executor, 很有可能是測試階段由于計算任務(wù)沒有被分配到錯誤的 Executor 節(jié)點而真正夜間計算時錯誤的 Executor 節(jié)點被分配了計算任務(wù)導(dǎo)致失敗.
  • 改進(jìn)的一種方案是: 代碼僅僅部署到 Scheduler 節(jié)點, 每次 Executor 拿到計算任務(wù)后, 通過 Scheduler 提供的 API 下載相應(yīng)的計算代碼.
  • 思考: 以上兩種方案解決的僅僅是計算代碼的部署問題, 沒有考慮計算代碼依賴的 Library 的問題. 如果計算代碼中有外部依賴, 比如一些 Native 的 Library, 就需要在每個 Executor 上安裝所有計算代碼依賴的 Library. 最悲催的是: 如果兩個計算任務(wù)依賴的 Library 的版本不一樣就悲催了.
• Web UI 模塊:
  • 查看 DAG,
  • 手動修復(fù)失敗任務(wù)
  • 看日志. 任務(wù)執(zhí)行日志查看.
    • 不支持 Tail 功能
    • 不支持 Grep 功能, 基本上可以斷定是: 不好用!

調(diào)度系統(tǒng)的需求明確了, 甚至各個模塊的功能都想清楚了, 我們就可以像"選妃子"一樣, 看市面上有哪些可選的開源項目

開源調(diào)度系統(tǒng)

• Linkedin 公司開源的 Azkaban
• Airbnb 公司開源的 Airflow

兩個系統(tǒng)都很不錯, 也能夠解決大多數(shù)需求. 網(wǎng)上針對兩個的比較也很多, 我就不班門弄斧了.

繼續(xù)上文中關(guān)于計算代碼依賴 Library 的思考

• 計算任務(wù)代碼可以分成兩部分:
  • 調(diào)度配置文件. 前置任務(wù), 啟動參數(shù)等
  • 計算代碼文件. 這才是會產(chǎn)生外部 Library 依賴的代碼.

因此, 如果我們把計算代碼裝進(jìn) Docker, 不就徹底解決了依賴問題?

-- EOF --