編者說：文章介紹了EasyScheduler的架構(gòu)設(shè)計，以及每個組件工作的方式。如果你的工作中遇到需要大數(shù)據(jù)工作流的場景可以嘗試使用這個架構(gòu)。由于文章比較長建議先收藏再閱讀。

Easy Scheduler 大數(shù)據(jù)工作流調(diào)度系統(tǒng)已經(jīng)開源，下載地址：https://github.com/analysys/

在對調(diào)度系統(tǒng)架構(gòu)說明之前，我們先來認識一下調(diào)度系統(tǒng)常用的名詞。

名詞解釋

• DAG：全稱Directed Acyclic Graph，簡稱DAG。工作流中的Task任務(wù)以有向無環(huán)圖的形式組裝起來，從入度為零的節(jié)點進行拓撲遍歷，直到無后繼節(jié)點為止。舉例如下圖 

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

系統(tǒng)架構(gòu) 

架構(gòu)說明

MasterServer

MasterServer采用分布式無中心設(shè)計理念，MasterServer主要負責(zé)DAG 任務(wù)切分、任務(wù)提交監(jiān)控，并同時監(jiān)聽其它MasterServer和WorkerServer的健康狀態(tài)。MasterServer服務(wù)啟動時向Zookeeper注冊臨時節(jié)點，通過監(jiān)聽Zookeeper臨時節(jié)點變化來進行容錯處理。

該服務(wù)內(nèi)主要包含:

• Distributed Quartz分布式調(diào)度組件，主要負責(zé)定時任務(wù)的啟停操作，當(dāng)quartz調(diào)起任務(wù)后，Master內(nèi)部會有線程池具體負責(zé)處理任務(wù)的后續(xù)操作
• MasterSchedulerThread是一個掃描線程，定時掃描數(shù)據(jù)庫中的 command 表，根據(jù)不同的 命令類型 進行不同的業(yè)務(wù)操作
• MasterExecThread主要是負責(zé)DAG任務(wù)切分、任務(wù)提交監(jiān)控、各種不同命令類型的邏輯處理
• MasterTaskExecThread主要負責(zé)任務(wù)的持久化

WorkerServer

WorkerServer也采用分布式無中心設(shè)計理念，WorkerServer主要負責(zé)任務(wù)的執(zhí)行和提供日志服務(wù)。WorkerServer服務(wù)啟動時向Zookeeper注冊臨時節(jié)點，并維持心跳。

該服務(wù)包含：

• FetchTaskThread主要負責(zé)不斷從 Task Queue 中領(lǐng)取任務(wù)，并根據(jù)不同任務(wù)類型調(diào)用 TaskScheduleThread 對應(yīng)執(zhí)行器。
• LoggerServer是一個RPC服務(wù)，提供日志分片查看、刷新和下載等功能

ZooKeeper

ZooKeeper服務(wù)，系統(tǒng)中的MasterServer和WorkerServer節(jié)點都通過ZooKeeper來進行集群管理和容錯。另外系統(tǒng)還基于ZooKeeper進行事件監(jiān)聽和分布式鎖。我們也曾經(jīng)基于Redis實現(xiàn)過隊列，不過我們希望EasyScheduler依賴到的組件盡量地少，所以***還是去掉了Redis實現(xiàn)。

Task Queue

提供任務(wù)隊列的操作，目前隊列也是基于Zookeeper來實現(xiàn)。由于隊列中存的信息較少，不必擔(dān)心隊列里數(shù)據(jù)過多的情況，實際上我們壓測過***數(shù)據(jù)存隊列，對系統(tǒng)穩(wěn)定性和性能沒影響。

Alert

提供告警相關(guān)接口，接口主要包括 告警 兩種類型的告警數(shù)據(jù)的存儲、查詢和通知功能。其中通知功能又有 郵件通知 和**SNMP(暫未實現(xiàn))**兩種。

API

API接口層，主要負責(zé)處理前端UI層的請求。該服務(wù)統(tǒng)一提供RESTful api向外部提供請求服務(wù)。接口包括工作流的創(chuàng)建、定義、查詢、修改、發(fā)布、下線、手工啟動、停止、暫停、恢復(fù)、從該節(jié)點開始執(zhí)行等等。

UI

系統(tǒng)的前端頁面，提供系統(tǒng)的各種可視化操作界面，詳見**系統(tǒng)使用手冊**部分。

• 架構(gòu)設(shè)計思想
• 去中心化vs中心化
• 中心化思想

中心化的設(shè)計理念比較簡單，分布式集群中的節(jié)點按照角色分工，大體上分為兩種角色： 

Master的角色主要負責(zé)任務(wù)分發(fā)并監(jiān)督Slave的健康狀態(tài)，可以動態(tài)的將任務(wù)均衡到Slave上，以致Slave節(jié)點不至于“忙死”或”閑死”的狀態(tài)。

Worker的角色主要負責(zé)任務(wù)的執(zhí)行工作并維護和Master的心跳，以便Master可以分配任務(wù)給Slave。

中心化思想設(shè)計存在的問題：

• 一旦Master出現(xiàn)了問題，則群龍無首，整個集群就會崩潰。為了解決這個問題，大多數(shù)Master/Slave架構(gòu)模式都采用了主備Master的設(shè)計方案，可以是熱備或者冷備，也可以是自動切換或手動切換，而且越來越多的新系統(tǒng)都開始具備自動選舉切換Master的能力,以提升系統(tǒng)的可用性。
• 另外一個問題是如果Scheduler在Master上，雖然可以支持一個DAG中不同的任務(wù)運行在不同的機器上，但是會產(chǎn)生Master的過負載。如果Scheduler在Slave上，則一個DAG中所有的任務(wù)都只能在某一臺機器上進行作業(yè)提交，則并行任務(wù)比較多的時候，Slave的壓力可能會比較大。

去中心化 

在去中心化設(shè)計里，通常沒有Master/Slave的概念，所有的角色都是一樣的，地位是平等的，全球互聯(lián)網(wǎng)就是一個典型的去中心化的分布式系統(tǒng)，聯(lián)網(wǎng)的任意節(jié)點設(shè)備down機，都只會影響很小范圍的功能。

去中心化設(shè)計的核心設(shè)計在于整個分布式系統(tǒng)中不存在一個區(qū)別于其他節(jié)點的”管理者”，因此不存在單點故障問題。但由于不存在” 管理者”節(jié)點所以每個節(jié)點都需要跟其他節(jié)點通信才得到必須要的機器信息，而分布式系統(tǒng)通信的不可靠行，則大大增加了上述功能的實現(xiàn)難度。

實際上，真正去中心化的分布式系統(tǒng)并不多見。反而動態(tài)中心化分布式系統(tǒng)正在不斷涌出。在這種架構(gòu)下，集群中的管理者是被動態(tài)選擇出來的，而不是預(yù)置的，并且集群在發(fā)生故障的時候，集群的節(jié)點會自發(fā)的舉行"會議"來選舉新的"管理者"去主持工作。最典型的案例就是ZooKeeper及Go語言實現(xiàn)的Etcd。

EasyScheduler的去中心化是Master/Worker注冊到Zookeeper中，實現(xiàn)Master集群和Worker集群無中心，并使用Zookeeper分布式鎖來選舉其中的一臺Master或Worker為“管理者”來執(zhí)行任務(wù)。

分布式鎖實踐

EasyScheduler使用ZooKeeper分布式鎖來實現(xiàn)同一時刻只有一臺Master執(zhí)行Scheduler，或者只有一臺Worker執(zhí)行任務(wù)的提交。

獲取分布式鎖的核心流程算法如下 

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

線程不足循環(huán)等待問題

如果一個DAG中沒有子流程，則如果Command中的數(shù)據(jù)條數(shù)大于線程池設(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來打破僵局，似乎有點差強人意，于是我們提出了以下三種方案來降低這種風(fēng)險：

• 計算所有Master的線程總和，然后對每一個DAG需要計算其需要的線程數(shù)，也就是在DAG流程執(zhí)行之前做預(yù)計算。因為是多Master線程池，所以總線程數(shù)不太可能實時獲取。
• 對單Master線程池進行判斷，如果線程池已經(jīng)滿了，則讓線程直接失敗。
• 增加一種資源不足的Command類型，如果線程池不足，則將主流程掛起。這樣線程池就有了新的線程，可以讓資源不足掛起的流程重新喚醒執(zhí)行。

注意：Master Scheduler線程在獲取Command的時候是FIFO的方式執(zhí)行的。

于是我們選擇了第三種方式來解決線程不足的問題。

容錯設(shè)計

容錯分為服務(wù)宕機容錯和任務(wù)重試，服務(wù)宕機容錯又分為Master容錯和Worker容錯兩種情況：

宕機容錯

服務(wù)容錯設(shè)計依賴于ZooKeeper的Watcher機制，實現(xiàn)原理如圖： 

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

Master容錯流程圖： 

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

Worker容錯流程圖： 

Master Scheduler線程一旦發(fā)現(xiàn)任務(wù)實例為” 需要容錯”狀態(tài)，則接管任務(wù)并進行重新提交。

注意：由于” 網(wǎng)絡(luò)抖動”可能會使得節(jié)點短時間內(nèi)失去和ZooKeeper的心跳，從而發(fā)生節(jié)點的remove事件。對于這種情況，我們使用最簡單的方式，那就是節(jié)點一旦和ZooKeeper發(fā)生超時連接，則直接將Master或Worker服務(wù)停掉。

任務(wù)失敗重試

這里首先要區(qū)分任務(wù)失敗重試、流程失敗恢復(fù)、流程失敗重跑的概念：

• 任務(wù)失敗重試是任務(wù)級別的，是調(diào)度系統(tǒng)自動進行的，比如一個Shell任務(wù)設(shè)置重試次數(shù)為3次，那么在Shell任務(wù)運行失敗后會自己再最多嘗試運行3次
• 流程失敗恢復(fù)是流程級別的，是手動進行的，恢復(fù)是從只能 從失敗的節(jié)點開始執(zhí)行 或 從當(dāng)前節(jié)點開始執(zhí)行
• 流程失敗重跑也是流程級別的，是手動進行的，重跑是從開始節(jié)點進行

接下來說正題，我們將工作流中的任務(wù)節(jié)點分了兩種類型。

• 一種是業(yè)務(wù)節(jié)點，這種節(jié)點都對應(yīng)一個實際的腳本或者處理語句，比如Shell節(jié)點，MR節(jié)點、Spark節(jié)點、依賴節(jié)點等。
• 還有一種是邏輯節(jié)點，這種節(jié)點不做實際的腳本或語句處理，只是整個流程流轉(zhuǎn)的邏輯處理，比如子流程節(jié)等。

每一個 業(yè)務(wù)節(jié)點 都可以配置失敗重試的次數(shù)，當(dāng)該任務(wù)節(jié)點失敗，會自動重試，直到成功或者超過配置的重試次數(shù)。邏輯節(jié)點 不支持失敗重試。但是邏輯節(jié)點里的任務(wù)支持重試。

如果工作流中有任務(wù)失敗達到***重試次數(shù)，工作流就會失敗停止，失敗的工作流可以手動進行重跑操作或者流程恢復(fù)操作

任務(wù)優(yōu)先級設(shè)計

在早期調(diào)度設(shè)計中，如果沒有優(yōu)先級設(shè)計，采用公平調(diào)度設(shè)計的話，會遇到先行提交的任務(wù)可能會和后繼提交的任務(wù)同時完成的情況，而不能做到設(shè)置流程或者任務(wù)的優(yōu)先級，因此我們對此進行了重新設(shè)計，目前我們設(shè)計如下：

• 按照 不同流程實例優(yōu)先級 優(yōu)先于 同一個流程實例優(yōu)先級 優(yōu)先于 同***程內(nèi)任務(wù)優(yōu)先級 優(yōu)先于 同***程內(nèi)任務(wù) 提交順序依次從高到低進行任務(wù)處理。

其中流程定義的優(yōu)先級是考慮到有些流程需要先于其他流程進行處理，這個可以在流程啟動或者定時啟動時配置，共有5級，依次為HIGHEST、HIGH、MEDIUM、LOW、LOWEST。如下圖

具體實現(xiàn)是根據(jù)任務(wù)實例的json解析優(yōu)先級，然后把 流程實例優(yōu)先級_流程實例id_任務(wù)優(yōu)先級_任務(wù)id 信息保存在ZooKeeper任務(wù)隊列中，當(dāng)從任務(wù)隊列獲取的時候，通過字符串比較即可得出最需要優(yōu)先執(zhí)行的任務(wù) 

任務(wù)的優(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的架構(gòu)原理及實現(xiàn)思路。