1.3.4　Flink作業調度

參與Flink程序執行的有多個進程，包括JobManager、TaskManager及JobClient，其中JobManager充當Master角色，TaskManager充當Worker角色。JobClient不是Flink任務執行過程的內部組件，而是執行過程的起始點。JobClient負責接收用戶提交的應用程序，創建對應的數據流，然后將數據流提交到JobManager上執行。一旦執行完畢，JobClient將執行結果發回給用戶。

Flink程序的執行過程如圖1-15所示。

Flink應用程序首先被提交到JobClient上，隨后JobClient將它提交到JobManager上，JobManager負責安排資源的分配和作業的執行。首先是資源的分配，然后將作業劃分為若干任務后提交到對應的TaskManager上。TaskManager在接收到任務后，初始化一個線程并開始執行程序。執行過程中TaskManager持續地將狀態的變化情況報告給JobManager，這些狀態包括開始執行（starting the execution）、正在執行（in progress）及完成（finished）。一旦作業的執行徹底完成，JobManager就將結果發回JobClient端。

Flink的分布式作業執行過程包含兩個重要的角色：Master和Worker，如圖1-16所示。

其中左側的JobManager部分即是Master，它負責管理整個集群中的資源并處理作業提交、作業監督；右側的兩個TaskManager則是Worker，這是工作進程，負責提供具體的資源并實際執行作業。

當用戶提交作業時，提交腳本會先啟動一個JobClient進程負責作業的編譯與提交。JobClient首先將用戶編寫的代碼編譯為一個JobGraph，在這個過程中，它還會進行一些檢查或優化等工作，例如判斷哪些運算符（算子）可以連接到同一個任務中，然后，JobClient將產生的JobGraph提交到集群中執行。

當作業提交到Dispatcher后，Dispatcher會首先啟動一個JobMaster組件，然后JobMaster會向ResourceManager申請資源來啟動作業中具體的任務。這時根據作業部署模式（關于作業部署模式，可參閱1.3.6節）的區別，TaskManager可能已經啟動或者尚未啟動。

（1）如果是前者，此時ResourceManager中已有記錄了TaskManager注冊的資源，可以直接選取空閑資源進行分配。

（2）否則ResourceManager首先需要向外部資源管理系統申請資源來啟動TaskManager，然后等待TaskManager注冊相應資源后再繼續選擇空閑資源進程分配。

目前Flink中TaskManager的資源是通過Slot來描述的，一個Slot一般可以執行一個具體的任務，但在一些情況下也可以執行多個相關聯的任務。ResourceManager選擇到空閑的Slot之后，就會通知相應的TaskManager“將該Slot分配給JobMaster XX”，然后TaskManager進行相應的記錄后會向JobManager進行注冊。JobManager收到TaskManager注冊的Slot后，就可以實際提交任務了。

TaskManager收到JobManager提交的任務之后會啟動一個新的線程來執行該任務。該任務啟動后就會開始進行預先指定的計算，并通過數據shuffle模塊互相交換數據。

基于1.3.3節中Slot管理和分配的邏輯，JobMaster負責維護作業中任務執行的狀態。如上文所述，客戶端會向JobMaster提交一個JobGraph，它代表了作業的邏輯結構。JobMaster會根據JobGraph按并發展開，從而得到JobMaster中關鍵的ExecutionGraph。ExecutionGraph的結構如圖1-17所示。

與JobGraph相比，ExecutionGraph中對于每個Task和中間結果等均創建了對應的對象，從而可以維護這些實體的信息與狀態。

在一個Flink Job中包含多個Task，因此另一個關鍵的問題是在Flink中按什么順序來調度這些Task。目前Flink提供了兩種基本的調度策略，即時延調度和即時調度，如圖1-18所示。

即時調度會在作業啟動時申請資源將所有的任務調度起來。這種調度算法主要用來調度可能沒有終止的流作業。與之對應，時延調度則是從源開始，按拓撲順序進行調度。簡單來講，時延調度會先調度沒有上游任務的源任務，當這些任務執行完成時，它會將輸出數據緩存到內存或者寫入磁盤中，然后，對于后續的任務，當它的前驅任務全部執行完成后，Flink就會將這些任務調度起來。這些任務會從讀取上游緩存的輸出數據進行自己的計算。這一過程繼續進行直到所有的任務完成計算。