關(guān)鍵詞：大數(shù)據(jù)；DAG有向無(wú)環(huán)圖；調(diào)度平臺(tái)

中圖法分類號(hào)：TP301 文獻(xiàn)標(biāo)識(shí)碼：A

1引言

大型商業(yè)運(yùn)營(yíng)階段生產(chǎn)的數(shù)據(jù)類型大多是傳統(tǒng)的結(jié)構(gòu)化數(shù)據(jù)。這些數(shù)據(jù)基本屬于隱私性和安全性等級(jí)十分高的貿(mào)易、商業(yè)、物流，以及保險(xiǎn)、股票等傳統(tǒng)支撐行業(yè)數(shù)據(jù)。而互聯(lián)網(wǎng)時(shí)代出現(xiàn)的數(shù)據(jù)類型大多屬于非結(jié)構(gòu)化的社交網(wǎng)絡(luò)數(shù)據(jù)、電子商務(wù)交易數(shù)據(jù)、圖片定位數(shù)據(jù)，以及商業(yè)智能報(bào)表、衛(wèi)星遙感數(shù)據(jù)、監(jiān)控錄像等非結(jié)構(gòu)化和二維碼像素?cái)?shù)據(jù)。因此，研究大數(shù)據(jù)任務(wù)流調(diào)度平臺(tái)，對(duì)于企業(yè)內(nèi)部自建大數(shù)據(jù)量的實(shí)時(shí)／離線同步、處理、清洗、治理、流程化、持久化等任務(wù)流程，具有重要的成本與運(yùn)營(yíng)意義。

2研究現(xiàn)狀

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

3大數(shù)據(jù)DAG任務(wù)流調(diào)度平臺(tái)架構(gòu)設(shè)計(jì)

本文詳細(xì)闡述了大數(shù)據(jù)DAG任務(wù)流調(diào)度平臺(tái)方案，即大數(shù)據(jù)DAG任務(wù)流調(diào)度平臺(tái)技術(shù)研究與應(yīng)用。整體架構(gòu)設(shè)計(jì)如圖1所示。

該框架在設(shè)計(jì)上充分考慮了大數(shù)據(jù)場(chǎng)景，利用去中心化的架構(gòu)，構(gòu)建整個(gè)調(diào)度集群，基于DAG有向無(wú)環(huán)圖，構(gòu)建整個(gè)任務(wù)流程體系，核心是使企業(yè)實(shí)時(shí)／離線大數(shù)據(jù)處理流程更加簡(jiǎn)易化，各組件模塊相互協(xié)作共同為此服務(wù)。其中用到的關(guān)鍵技術(shù)點(diǎn)將展開一一討論。

3.1協(xié)議設(shè)計(jì)

首先，需要進(jìn)行通信協(xié)議設(shè)計(jì)，傳統(tǒng)的http協(xié)議不滿足需求，我們需要高效、穩(wěn)定的通信協(xié)議，以解決通信中丟包、粘包、斷線重連、消息重發(fā)等問題。其次，進(jìn)行具體的通信協(xié)議設(shè)計(jì)，協(xié)議頭為15個(gè)字節(jié)。

該協(xié)議保障了傳輸?shù)姆€(wěn)定性和可擴(kuò)展性，Proto flag保障協(xié)議不被篡改，Real body size和Body size保證拆包、粘包、壓縮、加密的處理，Encrypt Flag可支持自定義協(xié)議加密算法等。

3.2任務(wù)引擎設(shè)計(jì)

首先需要說(shuō)明任務(wù)引擎在整個(gè)架構(gòu)里面的必要性和重要性，任務(wù)引擎為一個(gè)單位，可由不同程序不同模塊組成，甚至不同開發(fā)語(yǔ)言組成，這種架構(gòu)所設(shè)計(jì)的任務(wù)引擎具備很強(qiáng)的擴(kuò)展性和隔離性，如何建立它們之間的關(guān)系，是一個(gè)難題。其次探討任務(wù)引擎通信的設(shè)計(jì)，即如何做到不同結(jié)構(gòu)任務(wù)引擎之間的關(guān)聯(lián)。

每個(gè)任務(wù)引擎都有對(duì)外握手機(jī)制（輸入和輸出），要實(shí)現(xiàn)這一特性，必須定義任務(wù)引擎標(biāo)準(zhǔn)，每個(gè)任務(wù)引擎必須具備引擎名稱、輸入標(biāo)準(zhǔn)、輸出標(biāo)準(zhǔn)等標(biāo)準(zhǔn)信息，才能建立握手機(jī)制。之后，引擎之間就具備了信息交換、信息解析的能力，以使用Java語(yǔ)言編寫的sql引擎為例講解標(biāo)準(zhǔn)的定義，sql引擎應(yīng)具備接收數(shù)據(jù)源并且執(zhí)行動(dòng)作的能力。

3.3任務(wù)引擎熱加載機(jī)制

用戶可以進(jìn)行一個(gè)任務(wù)引擎（新引擎或迭代引擎）的上傳，當(dāng)上傳到worker時(shí)，worker會(huì)將用戶上傳的任務(wù)引擎做一致性校驗(yàn)（ md5/hash）.如發(fā)現(xiàn)此次上傳的引擎較舊引擎無(wú)變化，就不進(jìn)行處理，如有變化，則worker會(huì)將舊引擎（oldEngine）標(biāo)記為刪除狀態(tài)，將指針指向新引擎（newEngine），確保下一次任務(wù)使用新引擎，正在執(zhí)行的舊引擎會(huì)在它所有任務(wù)完成之后，從標(biāo)記刪除變?yōu)槲锢韯h除。

3.4 DAG結(jié)構(gòu)

調(diào)度系統(tǒng)需要使用DAG（有向無(wú)環(huán)圖）結(jié)構(gòu)，一般情況下，任務(wù)都是孤立的，任務(wù)之間也無(wú)關(guān)聯(lián)性可言，這樣的任務(wù)調(diào)度系統(tǒng)使用場(chǎng)景有限，因此無(wú)法實(shí)現(xiàn)任務(wù)順序性、任務(wù)關(guān)聯(lián)性、任務(wù)流程控制等，如何建立任務(wù)之間的關(guān)系，是一個(gè)難題?；贒AG結(jié)構(gòu)設(shè)計(jì)的任務(wù)流程，可以實(shí)現(xiàn)整個(gè)任務(wù)流程體系。

如圖2所示，總共有9個(gè)任務(wù)，每個(gè)任務(wù)都有關(guān)聯(lián)性和順序性，保證整個(gè)流程任務(wù)執(zhí)行的正確性是關(guān)鍵。

通過(guò)圖2可以看到，需要1個(gè)節(jié)點(diǎn)（ node）代表任務(wù)本身，每個(gè)節(jié)點(diǎn)存在多條邊（edge），每條邊存在前后2個(gè)節(jié)點(diǎn)（beforeNode，afterNode）。

整體執(zhí)行順序如圖3所示。

3.5資源介質(zhì)機(jī)制

任務(wù)引擎除了需要具備握手機(jī)制，還要保障其獨(dú)立性和可擴(kuò)展性。例如，某個(gè)引擎需要以文件內(nèi)容為輸入進(jìn)行解析，這時(shí)引擎如果要獲取這個(gè)文件，難度較大、效率較低，因?yàn)椴恢肋@個(gè)文件來(lái)源于哪里，此時(shí)需要有“人”幫任務(wù)引擎準(zhǔn)備好它所需要的“物資”，針對(duì)這種情況，我們提出資源介質(zhì)的概念。

用戶通過(guò)介質(zhì)人口進(jìn)行資源上傳，用戶不用關(guān)心當(dāng)前上傳的資源是什么類型的介質(zhì)，統(tǒng)一上傳到調(diào)度平臺(tái)，由調(diào)度平臺(tái)的資源介質(zhì)中心進(jìn)行管理分類等操作，當(dāng)任務(wù)引擎需要資源時(shí)，通過(guò)資源介質(zhì)出口到各個(gè)任務(wù)引擎，并自動(dòng)幫任務(wù)引擎準(zhǔn)備好這些資源。同樣，任務(wù)引擎也不關(guān)心當(dāng)前的資源介質(zhì)，直接使用即可。

3.6調(diào)度算法

本文實(shí)現(xiàn)的調(diào)度算法基于動(dòng)態(tài)負(fù)載均衡算法的變種，并基于幾個(gè)指標(biāo)來(lái)做決定，即內(nèi)存、cpu、任務(wù)數(shù)、線程數(shù)、系統(tǒng)負(fù)載、cpu負(fù)載，判斷當(dāng)前是否可以進(jìn)行調(diào)度，最終實(shí)現(xiàn)的計(jì)算式為：

其中，各字段的含義為ree為最終確定是否空閑可調(diào)度標(biāo)志free Thread為當(dāng)前系統(tǒng)空閑線程，cpu為cpu使用率，threshold為可配置閾值，mem為系統(tǒng)內(nèi)存占用，cpuLoad為系統(tǒng)cpu負(fù)載程度，systemLoad為系統(tǒng)整體負(fù)載程度。

當(dāng)freeThread大于1，cpu小于threshold，mem小于threshold，cpuLoad小于threshold，systemLoad小于threshold時(shí)，ree即為true，空閑狀態(tài)，可接受任務(wù)調(diào)度。

其中，cpu負(fù)載獲取算法為：

主要通過(guò)統(tǒng)計(jì)cpu rq上task處于runnable的平均時(shí)間。同時(shí)，根據(jù)不同周期，統(tǒng)計(jì)出不同的k線。其中，oldLoad為舊負(fù)載，newLoad為負(fù)載

系統(tǒng)負(fù)載獲取算法為（1分鐘）：

其中，old為舊負(fù)載，EXPi為l/exp（5 sec/1 min）固定點(diǎn)，F(xiàn)IXEDi為1《11固定點(diǎn)，new為新計(jì)算的負(fù)載。

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

3.7回調(diào)機(jī)制

調(diào)度系統(tǒng)回調(diào)機(jī)制十分重要。一般情況下，任務(wù)執(zhí)行完成后，需要通知任務(wù)發(fā)起者，告訴它任務(wù)成功還是失敗，面對(duì)這種情況，侵入式調(diào)度可以很好實(shí)現(xiàn)這一特性，調(diào)用當(dāng)前空間用戶定義的回調(diào)函數(shù)即可，但介于侵入式擴(kuò)展性問題，我們需要設(shè)計(jì)一個(gè)新的調(diào)度回調(diào)機(jī)制。

我們將解決以下問題：各個(gè)業(yè)務(wù)系統(tǒng)回調(diào)方式不一致，回調(diào)失敗的處理，回調(diào)性能。

為了讓回調(diào)機(jī)制不受限于某種類型，與業(yè)務(wù)系統(tǒng)剝離，我們?cè)O(shè)計(jì)了1個(gè)統(tǒng)一注冊(cè)回調(diào)接口，回調(diào)方式以插件形式注冊(cè)，支持更多回調(diào)方式進(jìn)行擴(kuò)展，讓用戶無(wú)需關(guān)注回調(diào)實(shí)現(xiàn)，只需要提交任務(wù)時(shí)進(jìn)行注冊(cè)回調(diào)邏輯，設(shè)計(jì)如下：用戶先自定義回調(diào)邏輯處理，然后提交任務(wù)，等待調(diào)度平臺(tái)任務(wù)執(zhí)行成功或失敗，任務(wù)執(zhí)行結(jié)束后會(huì)通過(guò)回調(diào)機(jī)制進(jìn)行回調(diào)，回調(diào)失敗會(huì)進(jìn)行重試，整個(gè)回調(diào)過(guò)程為異步操作，保證不阻塞業(yè)務(wù)，重試會(huì)有次數(shù)限制，當(dāng)達(dá)到對(duì)應(yīng)次數(shù)后，會(huì)等待一段時(shí)間進(jìn)行重試。

3.8信號(hào)機(jī)制

任務(wù)引擎會(huì)定義1個(gè)信號(hào)處理函數(shù)（標(biāo)準(zhǔn)），該處理函數(shù)處理當(dāng)前任務(wù)引擎需要釋放的資源，然后一般就做結(jié)束進(jìn)程exit處理，當(dāng)用戶觸發(fā)停止任務(wù).worker會(huì)發(fā)送1個(gè)中斷信號(hào)，系統(tǒng)空間會(huì)因當(dāng)前進(jìn)程信號(hào)中斷而調(diào)用系統(tǒng)內(nèi)核函數(shù)do_signal（）、handle_signal（）（linux），轉(zhuǎn)向任務(wù)引擎空間的信號(hào)處理函數(shù)（windows為SetConsoleCtrlHandler）.此時(shí)任務(wù)引擎進(jìn)行善后工作，信號(hào)處理函數(shù)結(jié)束后調(diào)用sigreturn（）進(jìn)行系統(tǒng)空間內(nèi)核善后工作，再返回任務(wù)引擎空間繼續(xù)執(zhí)行中斷前邏輯。

4效果評(píng)價(jià)

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

依托大數(shù)據(jù)DAG流程調(diào)度運(yùn)行過(guò)程的示例，流程配置采用非常簡(jiǎn)單的拖拽配置方式，任務(wù)執(zhí)行過(guò)程能按照順序執(zhí)行，并且整個(gè)執(zhí)行過(guò)程日志均有記錄，每個(gè)任務(wù)引擎也有自己的輸入／輸出標(biāo)準(zhǔn)，同時(shí)能拿到前置任務(wù)的輸出。整個(gè)平臺(tái)對(duì)使用人員而言非常人性化，只需在界面進(jìn)行配置，任務(wù)將會(huì)以預(yù)期的結(jié)果運(yùn)行。

綜合來(lái)看，本文研究的大數(shù)據(jù)DAG任務(wù)流調(diào)度平臺(tái)憑借其全方位、高擴(kuò)展、高性能的架構(gòu)設(shè)計(jì)，足以勝任企業(yè)內(nèi)部自建大數(shù)據(jù)量的實(shí)時(shí)／離線同步、處理等任務(wù)流程，大大提高了整體數(shù)據(jù)處理、任務(wù)調(diào)度能力。

5結(jié)束語(yǔ)

本文設(shè)計(jì)并實(shí)現(xiàn)了一套大數(shù)據(jù)DAG任務(wù)流調(diào)度平臺(tái)，并通過(guò)各個(gè)環(huán)節(jié)的設(shè)計(jì)，將高性能和高擴(kuò)展發(fā)揮極致，隨著互聯(lián)網(wǎng)的發(fā)展，有效解決了企業(yè)在現(xiàn)代數(shù)字化建設(shè)中，將新舊數(shù)據(jù)（通常為超大數(shù)據(jù)）處理整合并持續(xù)性擴(kuò)充的難題，通過(guò)多維度構(gòu)建起全方位的調(diào)度平臺(tái)，提高企業(yè)經(jīng)營(yíng)效率，節(jié)約大量人力，減少了企業(yè)在數(shù)據(jù)暴漲時(shí)代維護(hù)數(shù)據(jù)的投入。

作者簡(jiǎn)介：

許佳裕（1993—），大專，助理工程師，研究方向：大數(shù)據(jù)任務(wù)調(diào)度、數(shù)據(jù)清洗、框架設(shè)計(jì)建設(shè)、Lmux內(nèi)核。