Flink原理（四）——任務及調度

本文是博主閱讀官網文檔、博客及書籍後本身所思所得，如果存在有誤的地方，歡迎留言分享，謝謝！

---

1、任務調度

　　Flink是經過task slot的來定義執行資源的，爲優化資源的利用率，Flink經過slot共享，能夠將多個連續的task任務組成的一個pipeline放在一個slot中運行。當任務並行度>1時，並行任務中的每一個pipeline就會分配到一個slot去執行，這樣就會有一個問題，如果任務的並行度大於集羣中slot的個數了，會咋辦？首先，毫無疑問的一點是集羣中的slot中都會有pipeline在跑；其次，多的任務就會等待現有的運行結束再去運行。下面結合官網中提供的例子說明通常狀況下pipeline的分配狀況[1]。

　　下圖中，一個pipeline由Source - Map - Reduce組成，其中MapFunction的並行度爲4，ReduceFunction的並行度爲3，集羣有兩個TaskManager，其中每一個TaskManager有3個slot。

 

 　　圖中，每個pipeline由一個顏色表示，其中包含3個小圈，每個圈表明一個算子，ReduceFunction的並行度爲3，而MapFunction的爲4，因此從Map->Reduce會發生shuffer。圖中，任務會以ExecutionVertex 組成的 DAG 圖的形式分配到兩個TaskManage的slot中，在TaskManager2的slot中，運行在其中一個slot的DAG僅有兩個ExecutionVertex ，這裏會發生網絡shuffer。

 2、JobManager 數據結構

　　運行在各個TaskManager的slot中任務的調度是經過JobManager完成，除此以外，JobManager還負責失敗任務的重啓等。

　　當JobManager接受到JobGraph（JobGraph 是數據流的表現形式，包括JobVertex和中間結果IntermediateDataSet，每一個算子都有諸如並行度和執行代碼等屬性）會將其轉換爲ExecutionGraph，二者之間的關係以下圖所示：

　　對每一個 JobVertex，能夠當作是通過算子優化組成一個個operator chain（每一個operator chain能夠是一個或多個算子）和相關信息組成，而ExecutionVertex能夠看作是JobVertex的並行版，假設組成一個JobVertex的operator chain的並行度爲100，則在ExecutionGraph中，ExecutionVertex有100個，對應關係能夠多看看上圖。

　　在JobGraph轉換到ExecutionGraph的過程當中[2]，主要發生瞭如下轉變： 

• 加入了並行度的概念，成爲真正可調度的圖結構
• 生成了與JobVertex對應的ExecutionJobVertex，ExecutionVertex，與IntermediateDataSet對應的IntermediateResult和IntermediateResultPartition等，並行將經過這些類實現。

 每一個 ExecutionGraph 都有一個與其相關聯的做業狀態。此做業狀態指示做業執行的當前狀態，具體的狀態圖以下：

　　圖中各個狀態說明狀況很清楚，就不詳細說明，須要注意的是暫停狀態的做業將可能不會被徹底清理。暫停狀態（suspended）僅處於本地終止狀態，在Flink的HA模式下，意味着做業的執行僅在相應的 JobManager 上終止，但集羣的另外一個 JobManager 能夠從持久的HA存儲中恢復這個做業並從新啓動。

Ref：

[1]https://ci.apache.org/projects/flink/flink-docs-release-1.6/internals/job_scheduling.html

[2]http://www.javashuo.com/article/p-khdfegoi-mh.html