大數據DAG任務流調度平臺技術研究與應用

關鍵詞：大數據；DAG有向無環圖；調度平臺

中圖法分類號：TP301 文獻標識碼：A

1引言

大型商業運營階段生產的數據類型大多是傳統的結構化數據。這些數據基本屬于隱私性和安全性等級十分高的貿易、商業、物流，以及保險、股票等傳統支撐行業數據。而互聯網時代出現的數據類型大多屬于非結構化的社交網絡數據、電子商務交易數據、圖片定位數據，以及商業智能報表、衛星遙感數據、監控錄像等非結構化和二維碼像素數據。因此，研究大數據任務流調度平臺，對于企業內部自建大數據量的實時／離線同步、處理、清洗、治理、流程化、持久化等任務流程，具有重要的成本與運營意義。

2研究現狀

此前，傳統技術任務調度系統大多是侵入式調度，即需要依賴框架，若將框架拿掉或者換一個框架，則需要重新進行修改，同時部分調度系統雖然為非侵入式，但其機制的設計不足以承擔大數據時代數據的變化速度，不適應企業高速發展所需要的彈性、可定制性、獨立性?，F有的系統和技術已經無法解決當前大數據背景下企業數據量暴增的問題。

3大數據DAG任務流調度平臺架構設計

本文詳細闡述了大數據DAG任務流調度平臺方案，即大數據DAG任務流調度平臺技術研究與應用。整體架構設計如圖1所示。

該框架在設計上充分考慮了大數據場景，利用去中心化的架構，構建整個調度集群，基于DAG有向無環圖，構建整個任務流程體系，核心是使企業實時／離線大數據處理流程更加簡易化，各組件模塊相互協作共同為此服務。其中用到的關鍵技術點將展開一一討論。

3.1協議設計

首先，需要進行通信協議設計，傳統的http協議不滿足需求，我們需要高效、穩定的通信協議，以解決通信中丟包、粘包、斷線重連、消息重發等問題。其次，進行具體的通信協議設計，協議頭為15個字節。

該協議保障了傳輸的穩定性和可擴展性，Proto flag保障協議不被篡改，Real body size和Body size保證拆包、粘包、壓縮、加密的處理，Encrypt Flag可支持自定義協議加密算法等。

3.2任務引擎設計

首先需要說明任務引擎在整個架構里面的必要性和重要性，任務引擎為一個單位，可由不同程序不同模塊組成，甚至不同開發語言組成，這種架構所設計的任務引擎具備很強的擴展性和隔離性，如何建立它們之間的關系，是一個難題。其次探討任務引擎通信的設計，即如何做到不同結構任務引擎之間的關聯。

每個任務引擎都有對外握手機制（輸入和輸出），要實現這一特性，必須定義任務引擎標準，每個任務引擎必須具備引擎名稱、輸入標準、輸出標準等標準信息，才能建立握手機制。之后，引擎之間就具備了信息交換、信息解析的能力，以使用Java語言編寫的sql引擎為例講解標準的定義，sql引擎應具備接收數據源并且執行動作的能力。

3.3任務引擎熱加載機制

用戶可以進行一個任務引擎（新引擎或迭代引擎）的上傳，當上傳到worker時，worker會將用戶上傳的任務引擎做一致性校驗（ md5/hash）.如發現此次上傳的引擎較舊引擎無變化，就不進行處理，如有變化，則worker會將舊引擎（oldEngine）標記為刪除狀態，將指針指向新引擎（newEngine），確保下一次任務使用新引擎，正在執行的舊引擎會在它所有任務完成之后，從標記刪除變為物理刪除。

3.4 DAG結構

調度系統需要使用DAG（有向無環圖）結構，一般情況下，任務都是孤立的，任務之間也無關聯性可言，這樣的任務調度系統使用場景有限，因此無法實現任務順序性、任務關聯性、任務流程控制等，如何建立任務之間的關系，是一個難題?；贒AG結構設計的任務流程，可以實現整個任務流程體系。

如圖2所示，總共有9個任務，每個任務都有關聯性和順序性，保證整個流程任務執行的正確性是關鍵。

通過圖2可以看到，需要1個節點（ node）代表任務本身，每個節點存在多條邊（edge），每條邊存在前后2個節點（beforeNode，afterNode）。

整體執行順序如圖3所示。

3.5資源介質機制

任務引擎除了需要具備握手機制，還要保障其獨立性和可擴展性。例如，某個引擎需要以文件內容為輸入進行解析，這時引擎如果要獲取這個文件，難度較大、效率較低，因為不知道這個文件來源于哪里，此時需要有“人”幫任務引擎準備好它所需要的“物資”，針對這種情況，我們提出資源介質的概念。

用戶通過介質人口進行資源上傳，用戶不用關心當前上傳的資源是什么類型的介質，統一上傳到調度平臺，由調度平臺的資源介質中心進行管理分類等操作，當任務引擎需要資源時，通過資源介質出口到各個任務引擎，并自動幫任務引擎準備好這些資源。同樣，任務引擎也不關心當前的資源介質，直接使用即可。

3.6調度算法

本文實現的調度算法基于動態負載均衡算法的變種，并基于幾個指標來做決定，即內存、cpu、任務數、線程數、系統負載、cpu負載，判斷當前是否可以進行調度，最終實現的計算式為：

其中，各字段的含義為ree為最終確定是否空閑可調度標志free Thread為當前系統空閑線程，cpu為cpu使用率，threshold為可配置閾值，mem為系統內存占用，cpuLoad為系統cpu負載程度，systemLoad為系統整體負載程度。

當freeThread大于1，cpu小于threshold，mem小于threshold，cpuLoad小于threshold，systemLoad小于threshold時，ree即為true，空閑狀態，可接受任務調度。

其中，cpu負載獲取算法為：

主要通過統計cpu rq上task處于runnable的平均時間。同時，根據不同周期，統計出不同的k線。其中，oldLoad為舊負載，newLoad為負載

系統負載獲取算法為（1分鐘）：

其中，old為舊負載，EXPi為l/exp（5 sec/1 min）固定點，FIXEDi為1《11固定點，new為新計算的負載。

至于5分鐘和15分鐘的計算，將式（4）中的EXP₁換成EXP₅/EXP₁₅即可。

3.7回調機制

調度系統回調機制十分重要。一般情況下，任務執行完成后，需要通知任務發起者，告訴它任務成功還是失敗，面對這種情況，侵入式調度可以很好實現這一特性，調用當前空間用戶定義的回調函數即可，但介于侵入式擴展性問題，我們需要設計一個新的調度回調機制。

我們將解決以下問題：各個業務系統回調方式不一致，回調失敗的處理，回調性能。

為了讓回調機制不受限于某種類型，與業務系統剝離，我們設計了1個統一注冊回調接口，回調方式以插件形式注冊，支持更多回調方式進行擴展，讓用戶無需關注回調實現，只需要提交任務時進行注冊回調邏輯，設計如下：用戶先自定義回調邏輯處理，然后提交任務，等待調度平臺任務執行成功或失敗，任務執行結束后會通過回調機制進行回調，回調失敗會進行重試，整個回調過程為異步操作，保證不阻塞業務，重試會有次數限制，當達到對應次數后，會等待一段時間進行重試。

3.8信號機制

任務引擎會定義1個信號處理函數（標準），該處理函數處理當前任務引擎需要釋放的資源，然后一般就做結束進程exit處理，當用戶觸發停止任務.worker會發送1個中斷信號，系統空間會因當前進程信號中斷而調用系統內核函數do_signal（）、handle_signal（）（linux），轉向任務引擎空間的信號處理函數（windows為SetConsoleCtrlHandler）.此時任務引擎進行善后工作，信號處理函數結束后調用sigreturn（）進行系統空間內核善后工作，再返回任務引擎空間繼續執行中斷前邏輯。

4效果評價

按照上文所述的大數據DAG任務流調度平臺方案，我們進行了編碼實現，主要分為用戶端流程任務配置、大數據端dag流程任務調度、任務進度、任務日志追溯等功能，這里采用的是B/S前后端分離，去中心化架構模式，通過集群負載均衡的方式，支持超大數據的任務調度，并支持任務引擎動態擴展，處理大數據處理過程中非常復雜的任務依賴關系，致力于實現在離線、實時超大量任務流程中的高性能調度和穩定性。另外，我們還實現了告警和消息通知機制，能在任務異常時，及時告知、處理，以挽回損失。

依托大數據DAG流程調度運行過程的示例，流程配置采用非常簡單的拖拽配置方式，任務執行過程能按照順序執行，并且整個執行過程日志均有記錄，每個任務引擎也有自己的輸入／輸出標準，同時能拿到前置任務的輸出。整個平臺對使用人員而言非常人性化，只需在界面進行配置，任務將會以預期的結果運行。

綜合來看，本文研究的大數據DAG任務流調度平臺憑借其全方位、高擴展、高性能的架構設計，足以勝任企業內部自建大數據量的實時／離線同步、處理等任務流程，大大提高了整體數據處理、任務調度能力。

5結束語

本文設計并實現了一套大數據DAG任務流調度平臺，并通過各個環節的設計，將高性能和高擴展發揮極致，隨著互聯網的發展，有效解決了企業在現代數字化建設中，將新舊數據（通常為超大數據）處理整合并持續性擴充的難題，通過多維度構建起全方位的調度平臺，提高企業經營效率，節約大量人力，減少了企業在數據暴漲時代維護數據的投入。

作者簡介：

許佳裕（1993—），大專，助理工程師，研究方向：大數據任務調度、數據清洗、框架設計建設、Lmux內核。