怎么做大數(shù)據(jù)工作流調(diào)度系統(tǒng)?大廠架構師一語點破!
編者說:文章介紹了EasyScheduler的架構設計,以及每個組件工作的方式。如果你的工作中遇到需要大數(shù)據(jù)工作流的場景可以嘗試使用這個架構。由于文章比較長建議先收藏再閱讀。
Easy Scheduler 大數(shù)據(jù)工作流調(diào)度系統(tǒng)已經(jīng)開源,下載地址:https://github.com/analysys/
在對調(diào)度系統(tǒng)架構說明之前,我們先來認識一下調(diào)度系統(tǒng)常用的名詞。
名詞解釋
- DAG:全稱Directed Acyclic Graph,簡稱DAG。工作流中的Task任務以有向無環(huán)圖的形式組裝起來,從入度為零的節(jié)點進行拓撲遍歷,直到無后繼節(jié)點為止。舉例如下圖

- 流程定義:通過拖拽任務節(jié)點并建立任務節(jié)點的關聯(lián)所形成的可視化 DAG
- 流程實例:流程實例是流程定義的實例化,可以通過手動啟動或定時調(diào)度生成
- 任務實例:任務實例是流程定義中任務節(jié)點的實例化,標識著具體的任務執(zhí)行狀態(tài)
- 任務類型:目前支持有SHELL、SQL、SUB_PROCESS、PROCEDURE、MR、SPARK、PYTHON、DEPENDENT,同時計劃支持動態(tài)插件擴展,注意:其中子 SUB_PROCESS 也是一個單獨的流程定義,是可以單獨啟動執(zhí)行的
- 調(diào)度方式:系統(tǒng)支持基于cron表達式的定時調(diào)度和手動調(diào)度。命令類型支持:啟動工作流、從當前節(jié)點開始執(zhí)行、恢復被容錯的工作流、恢復暫停流程、從失敗節(jié)點開始執(zhí)行、補數(shù)、調(diào)度、重跑、暫停、停止、恢復等待線程。其中 恢復被容錯的工作流 和 恢復等待線程 兩種命令類型是由調(diào)度內(nèi)部控制使用,外部無法調(diào)用
- 定時調(diào)度:系統(tǒng)采用 quartz 分布式調(diào)度器,并同時支持cron表達式可視化的生成
- 依賴:系統(tǒng)不單單支持 DAG 簡單的前驅(qū)和后繼節(jié)點之間的依賴,同時還提供 任務依賴 節(jié)點,支持 流程間的自定義任務依賴
- 優(yōu)先級:支持流程實例和任務實例的優(yōu)先級,如果流程實例和任務實例的優(yōu)先級不設置,則默認是先進先出
- 郵件告警:支持 SQL任務 查詢結(jié)果郵件發(fā)送,流程實例運行結(jié)果郵件告警及容錯告警通知
- 失敗策略:對于并行運行的任務,如果有任務失敗,提供兩種失敗策略處理方式, 繼續(xù) 是指不管并行運行任務的狀態(tài),直到流程失敗結(jié)束。結(jié)束 是指一旦發(fā)現(xiàn)失敗任務,則同時Kill掉正在運行的并行任務,流程失敗結(jié)束
- 補數(shù):補歷史數(shù)據(jù),支持 區(qū)間并行和串行 兩種補數(shù)方式
系統(tǒng)架構

架構說明
MasterServer
MasterServer采用分布式無中心設計理念,MasterServer主要負責DAG 任務切分、任務提交監(jiān)控,并同時監(jiān)聽其它MasterServer和WorkerServer的健康狀態(tài)。MasterServer服務啟動時向Zookeeper注冊臨時節(jié)點,通過監(jiān)聽Zookeeper臨時節(jié)點變化來進行容錯處理。
該服務內(nèi)主要包含:
- Distributed Quartz分布式調(diào)度組件,主要負責定時任務的啟停操作,當quartz調(diào)起任務后,Master內(nèi)部會有線程池具體負責處理任務的后續(xù)操作
- MasterSchedulerThread是一個掃描線程,定時掃描數(shù)據(jù)庫中的 command 表,根據(jù)不同的 命令類型 進行不同的業(yè)務操作
- MasterExecThread主要是負責DAG任務切分、任務提交監(jiān)控、各種不同命令類型的邏輯處理
- MasterTaskExecThread主要負責任務的持久化
WorkerServer
WorkerServer也采用分布式無中心設計理念,WorkerServer主要負責任務的執(zhí)行和提供日志服務。WorkerServer服務啟動時向Zookeeper注冊臨時節(jié)點,并維持心跳。
該服務包含:
- FetchTaskThread主要負責不斷從 Task Queue 中領取任務,并根據(jù)不同任務類型調(diào)用 TaskScheduleThread 對應執(zhí)行器。
- LoggerServer是一個RPC服務,提供日志分片查看、刷新和下載等功能
ZooKeeper
ZooKeeper服務,系統(tǒng)中的MasterServer和WorkerServer節(jié)點都通過ZooKeeper來進行集群管理和容錯。另外系統(tǒng)還基于ZooKeeper進行事件監(jiān)聽和分布式鎖。我們也曾經(jīng)基于Redis實現(xiàn)過隊列,不過我們希望EasyScheduler依賴到的組件盡量地少,所以***還是去掉了Redis實現(xiàn)。
Task Queue
提供任務隊列的操作,目前隊列也是基于Zookeeper來實現(xiàn)。由于隊列中存的信息較少,不必擔心隊列里數(shù)據(jù)過多的情況,實際上我們壓測過***數(shù)據(jù)存隊列,對系統(tǒng)穩(wěn)定性和性能沒影響。
Alert
提供告警相關接口,接口主要包括 告警 兩種類型的告警數(shù)據(jù)的存儲、查詢和通知功能。其中通知功能又有 郵件通知 和**SNMP(暫未實現(xiàn))**兩種。
API
API接口層,主要負責處理前端UI層的請求。該服務統(tǒng)一提供RESTful api向外部提供請求服務。接口包括工作流的創(chuàng)建、定義、查詢、修改、發(fā)布、下線、手工啟動、停止、暫停、恢復、從該節(jié)點開始執(zhí)行等等。
UI
系統(tǒng)的前端頁面,提供系統(tǒng)的各種可視化操作界面,詳見**系統(tǒng)使用手冊**部分。
- 架構設計思想
- 去中心化vs中心化
- 中心化思想
中心化的設計理念比較簡單,分布式集群中的節(jié)點按照角色分工,大體上分為兩種角色:

Master的角色主要負責任務分發(fā)并監(jiān)督Slave的健康狀態(tài),可以動態(tài)的將任務均衡到Slave上,以致Slave節(jié)點不至于“忙死”或”閑死”的狀態(tài)。
Worker的角色主要負責任務的執(zhí)行工作并維護和Master的心跳,以便Master可以分配任務給Slave。
中心化思想設計存在的問題:
- 一旦Master出現(xiàn)了問題,則群龍無首,整個集群就會崩潰。為了解決這個問題,大多數(shù)Master/Slave架構模式都采用了主備Master的設計方案,可以是熱備或者冷備,也可以是自動切換或手動切換,而且越來越多的新系統(tǒng)都開始具備自動選舉切換Master的能力,以提升系統(tǒng)的可用性。
- 另外一個問題是如果Scheduler在Master上,雖然可以支持一個DAG中不同的任務運行在不同的機器上,但是會產(chǎn)生Master的過負載。如果Scheduler在Slave上,則一個DAG中所有的任務都只能在某一臺機器上進行作業(yè)提交,則并行任務比較多的時候,Slave的壓力可能會比較大。
去中心化

在去中心化設計里,通常沒有Master/Slave的概念,所有的角色都是一樣的,地位是平等的,全球互聯(lián)網(wǎng)就是一個典型的去中心化的分布式系統(tǒng),聯(lián)網(wǎng)的任意節(jié)點設備down機,都只會影響很小范圍的功能。
去中心化設計的核心設計在于整個分布式系統(tǒng)中不存在一個區(qū)別于其他節(jié)點的”管理者”,因此不存在單點故障問題。但由于不存在” 管理者”節(jié)點所以每個節(jié)點都需要跟其他節(jié)點通信才得到必須要的機器信息,而分布式系統(tǒng)通信的不可靠行,則大大增加了上述功能的實現(xiàn)難度。
實際上,真正去中心化的分布式系統(tǒng)并不多見。反而動態(tài)中心化分布式系統(tǒng)正在不斷涌出。在這種架構下,集群中的管理者是被動態(tài)選擇出來的,而不是預置的,并且集群在發(fā)生故障的時候,集群的節(jié)點會自發(fā)的舉行"會議"來選舉新的"管理者"去主持工作。最典型的案例就是ZooKeeper及Go語言實現(xiàn)的Etcd。
EasyScheduler的去中心化是Master/Worker注冊到Zookeeper中,實現(xiàn)Master集群和Worker集群無中心,并使用Zookeeper分布式鎖來選舉其中的一臺Master或Worker為“管理者”來執(zhí)行任務。
分布式鎖實踐
EasyScheduler使用ZooKeeper分布式鎖來實現(xiàn)同一時刻只有一臺Master執(zhí)行Scheduler,或者只有一臺Worker執(zhí)行任務的提交。
獲取分布式鎖的核心流程算法如下

EasyScheduler中Scheduler線程分布式鎖實現(xiàn)流程圖:

線程不足循環(huán)等待問題
如果一個DAG中沒有子流程,則如果Command中的數(shù)據(jù)條數(shù)大于線程池設置的閾值,則直接流程等待或失敗。
如果一個大的DAG中嵌套了很多子流程,如下圖則會產(chǎn)生“死等”狀態(tài):

上圖中MainFlowThread等待SubFlowThread1結(jié)束,SubFlowThread1等待SubFlowThread2結(jié)束,SubFlowThread2等待SubFlowThread3結(jié)束,而SubFlowThread3等待線程池有新線程,則整個DAG流程不能結(jié)束,從而其中的線程也不能釋放。這樣就形成的子父流程循環(huán)等待的狀態(tài)。此時除非啟動新的Master來增加線程來打破這樣的”僵局”,否則調(diào)度集群將不能再使用。
對于啟動新Master來打破僵局,似乎有點差強人意,于是我們提出了以下三種方案來降低這種風險:
- 計算所有Master的線程總和,然后對每一個DAG需要計算其需要的線程數(shù),也就是在DAG流程執(zhí)行之前做預計算。因為是多Master線程池,所以總線程數(shù)不太可能實時獲取。
- 對單Master線程池進行判斷,如果線程池已經(jīng)滿了,則讓線程直接失敗。
- 增加一種資源不足的Command類型,如果線程池不足,則將主流程掛起。這樣線程池就有了新的線程,可以讓資源不足掛起的流程重新喚醒執(zhí)行。
注意:Master Scheduler線程在獲取Command的時候是FIFO的方式執(zhí)行的。
于是我們選擇了第三種方式來解決線程不足的問題。
容錯設計
容錯分為服務宕機容錯和任務重試,服務宕機容錯又分為Master容錯和Worker容錯兩種情況:
宕機容錯
服務容錯設計依賴于ZooKeeper的Watcher機制,實現(xiàn)原理如圖:

其中Master監(jiān)控其他Master和Worker的目錄,如果監(jiān)聽到remove事件,則會根據(jù)具體的業(yè)務邏輯進行流程實例容錯或者任務實例容錯。
Master容錯流程圖:

ZooKeeper Master容錯完成之后則重新由EasyScheduler中Scheduler線程調(diào)度,遍歷DAG 找到”正在運行”和“提交成功”的任務,對”正在運行”的任務監(jiān)控其任務實例的狀態(tài),對”提交成功”的任務需要判斷Task Queue中是否已經(jīng)存在,如果存在則同樣監(jiān)控任務實例的狀態(tài),如果不存在則重新提交任務實例。
Worker容錯流程圖:

Master Scheduler線程一旦發(fā)現(xiàn)任務實例為” 需要容錯”狀態(tài),則接管任務并進行重新提交。
注意:由于” 網(wǎng)絡抖動”可能會使得節(jié)點短時間內(nèi)失去和ZooKeeper的心跳,從而發(fā)生節(jié)點的remove事件。對于這種情況,我們使用最簡單的方式,那就是節(jié)點一旦和ZooKeeper發(fā)生超時連接,則直接將Master或Worker服務停掉。
任務失敗重試
這里首先要區(qū)分任務失敗重試、流程失敗恢復、流程失敗重跑的概念:
- 任務失敗重試是任務級別的,是調(diào)度系統(tǒng)自動進行的,比如一個Shell任務設置重試次數(shù)為3次,那么在Shell任務運行失敗后會自己再最多嘗試運行3次
- 流程失敗恢復是流程級別的,是手動進行的,恢復是從只能 從失敗的節(jié)點開始執(zhí)行 或 從當前節(jié)點開始執(zhí)行
- 流程失敗重跑也是流程級別的,是手動進行的,重跑是從開始節(jié)點進行
接下來說正題,我們將工作流中的任務節(jié)點分了兩種類型。
- 一種是業(yè)務節(jié)點,這種節(jié)點都對應一個實際的腳本或者處理語句,比如Shell節(jié)點,MR節(jié)點、Spark節(jié)點、依賴節(jié)點等。
- 還有一種是邏輯節(jié)點,這種節(jié)點不做實際的腳本或語句處理,只是整個流程流轉(zhuǎn)的邏輯處理,比如子流程節(jié)等。
每一個 業(yè)務節(jié)點 都可以配置失敗重試的次數(shù),當該任務節(jié)點失敗,會自動重試,直到成功或者超過配置的重試次數(shù)。邏輯節(jié)點 不支持失敗重試。但是邏輯節(jié)點里的任務支持重試。
如果工作流中有任務失敗達到***重試次數(shù),工作流就會失敗停止,失敗的工作流可以手動進行重跑操作或者流程恢復操作
任務優(yōu)先級設計
在早期調(diào)度設計中,如果沒有優(yōu)先級設計,采用公平調(diào)度設計的話,會遇到先行提交的任務可能會和后繼提交的任務同時完成的情況,而不能做到設置流程或者任務的優(yōu)先級,因此我們對此進行了重新設計,目前我們設計如下:
- 按照 不同流程實例優(yōu)先級 優(yōu)先于 同一個流程實例優(yōu)先級 優(yōu)先于 同***程內(nèi)任務優(yōu)先級 優(yōu)先于 同***程內(nèi)任務 提交順序依次從高到低進行任務處理。
其中流程定義的優(yōu)先級是考慮到有些流程需要先于其他流程進行處理,這個可以在流程啟動或者定時啟動時配置,共有5級,依次為HIGHEST、HIGH、MEDIUM、LOW、LOWEST。如下圖
具體實現(xiàn)是根據(jù)任務實例的json解析優(yōu)先級,然后把 流程實例優(yōu)先級_流程實例id_任務優(yōu)先級_任務id 信息保存在ZooKeeper任務隊列中,當從任務隊列獲取的時候,通過字符串比較即可得出最需要優(yōu)先執(zhí)行的任務

任務的優(yōu)先級也分為5級,依次為HIGHEST、HIGH、MEDIUM、LOW、LOWEST。如下圖

Logback和gRPC實現(xiàn)日志訪問
由于Web(UI)和Worker不一定在同一臺機器上,所以查看日志不能像查詢本地文件那樣。有兩種方案:
- 將日志放到ES搜索引擎上
- 通過gRPC通信獲取遠程日志信息
介于考慮到盡可能的EasyScheduler的輕量級性,所以選擇了gRPC實現(xiàn)遠程訪問日志信息。

總結(jié)
本文從調(diào)度出發(fā),初步介紹了大數(shù)據(jù)分布式工作流調(diào)度系統(tǒng)--EasyScheduler的架構原理及實現(xiàn)思路。