作者：岳猛
整理：平凡的世界-zkx
本文根據(jù) Apache Flink 系列直播課程整理而成，由 Apache Flink Contributor、網(wǎng)易云音樂實時計算平臺研發(fā)工程師岳猛分享。主要分享內(nèi)容為 Flink Job 執(zhí)行作業(yè)的流程，文章將從兩個方面進行分享：一是如何從 Program 到物理執(zhí)行計劃，二是生成物理執(zhí)行計劃后該如何調(diào)度和執(zhí)行。
Flink 四層轉(zhuǎn)化流程
Flink 有四層轉(zhuǎn)換流程，第一層為 Program 到 StreamGraph；第二層為 StreamGraph 到 JobGraph；第三層為 JobGraph 到 ExecutionGraph；第四層為 ExecutionGraph 到物理執(zhí)行計劃。通過對 Program 的執(zhí)行，能夠生成一個 DAG 執(zhí)行圖，即邏輯執(zhí)行圖。如下：
image.png
第一部分將先講解四層轉(zhuǎn)化的流程，然后將以詳細案例講解四層的具體轉(zhuǎn)化。
- 第一層 StreamGraph 從 Source 節(jié)點開始，每一次 transform 生成一個 StreamNode，兩個 StreamNode 通過 StreamEdge 連接在一起,形成 StreamNode 和 StreamEdge 構(gòu)成的DAG。
- 第二層 JobGraph，依舊從 Source 節(jié)點開始，然后去遍歷尋找能夠嵌到一起的 operator，如果能夠嵌到一起則嵌到一起，不能嵌到一起的單獨生成 jobVertex，通過 JobEdge 鏈接上下游 JobVertex，最終形成 JobVertex 層面的 DAG。
- JobVertex DAG 提交到任務(wù)以后，從 Source 節(jié)點開始排序,根據(jù) JobVertex 生成ExecutionJobVertex，根據(jù) jobVertex的IntermediateDataSet 構(gòu)建IntermediateResult，然后 IntermediateResult 構(gòu)建上下游的依賴關(guān)系，形成 ExecutionJobVertex 層面的 DAG 即 ExecutionGraph。
- 最后通過 ExecutionGraph 層到物理執(zhí)行層。
Program 到 StreamGraph 的轉(zhuǎn)化
Program 轉(zhuǎn)換成 StreamGraph 具體分為三步：
- 從 StreamExecutionEnvironment.execute 開始執(zhí)行程序，將 transform 添加到 StreamExecutionEnvironment 的 transformations。
- 調(diào)用 StreamGraphGenerator 的 generateInternal 方法，遍歷 transformations 構(gòu)建 StreamNode 及 StreamEage。
通過 StreamEdge 連接 StreamNode。
image.png
通過 WindowWordCount 來看代碼到 StreamGraph 的轉(zhuǎn)化，在 flatMap transform 設(shè)置 slot 共享組為 flatMap_sg，并發(fā)設(shè)置為 4，在聚合的操作中設(shè)置 slot 共享組為 sum_sg， sum() 和 counts() 并發(fā)設(shè)置為 3，這樣設(shè)置主要是為了演示后面如何嵌到一起的，跟上下游節(jié)點的并發(fā)以及上游的共享組有關(guān)。
WindowWordCount 代碼中可以看到，在 readTextFile() 中會生成一個 transform，且 transform 的 ID 是 1；然后到 flatMap() 會生成一個 transform， transform 的 ID 是 2；接著到 keyBy() 生成一個 transform 的 ID 是 3；再到 sum() 生成一個 transform 的 ID 是 4；最后到 counts()生成 transform 的 ID 是 5。
image.png
transform 的結(jié)構(gòu)如圖所示，第一個是 flatMap 的 transform，第二個是 window 的 transform，第三個是 SinkTransform 的 transform。除此之外，還能在 transform 的結(jié)構(gòu)中看到每個 transform 的 input 是什么。
接下來介紹一下 StreamNode 和 StreamEdge。
- StreamNode 是用來描述 operator 的邏輯節(jié)點，其關(guān)鍵成員變量有 slotSharingGroup、jobVertexClass、inEdges、outEdges以及transformationUID；
StreamEdge 是用來描述兩個 operator 邏輯的鏈接邊，其關(guān)鍵變量有 sourceVertex、targetVertex。
image.png
WindowWordCount transform 到 StreamGraph 轉(zhuǎn)化如圖所示，StreamExecutionEnvironment 的 transformations 存在 3 個 transform，分別是 Flat Map（Id 2）、Window（Id 4）、Sink（Id 5）。
transform 的時候首先遞歸處理 transform 的 input，生成 StreamNode，然后通過 StreamEdge 鏈接上下游 StreamNode。需要注意的是，有些 transform 操作并不會生成StreamNode 如 PartitionTransformtion，而是生成個虛擬節(jié)點。
image.png
在轉(zhuǎn)換完成后可以看到，streamNodes 有四種 transform 形式，分別為 Source、Flat Map、Window、Sink。
image.png
每個 streamNode 對象都攜帶并發(fā)個數(shù)、slotSharingGroup、執(zhí)行類等運行信息。
StreamGraph 到 JobGraph 的轉(zhuǎn)化
Flink Job 執(zhí)行流程
Flink On Yarn 模式
基于 Yarn 層面的架構(gòu)類似 Spark on Yarn 模式，都是由 Client 提交 App 到 RM 上面去運行，然后 RM 分配第一個 container 去運行 AM，然后由 AM 去負責資源的監(jiān)督和管理。需要說明的是，F(xiàn)link 的 Yarn 模式更加類似 Spark on Yarn 的 cluster 模式，在 cluster 模式中，dirver 將作為 AM 中的一個線程去運行。Flink on Yarn 模式也是會將 JobManager 啟動在 container 里面，去做個 driver 類似的任務(wù)調(diào)度和分配，Yarn AM 與 Flink JobManager 在同一個 Container 中，這樣 AM 可以知道 Flink JobManager 的地址，從而 AM 可以申請 Container 去啟動 Flink TaskManager。待 Flink 成功運行在 Yarn 集群上，F(xiàn)link Yarn Client 就可以提交 Flink Job 到 Flink JobManager，并進行后續(xù)的映射、調(diào)度和計算處理。
Fink on Yarn 的缺陷
資源分配是靜態(tài)的，一個作業(yè)需要在啟動時獲取所需的資源并且在它的生命周期里一直持有這些資源。這導致了作業(yè)不能隨負載變化而動態(tài)調(diào)整，在負載下降時無法歸還空閑的資源，在負載上升時也無法動態(tài)擴展。
On-Yarn 模式下，所有的 container 都是固定大小的，導致無法根據(jù)作業(yè)需求來調(diào)整 container 的結(jié)構(gòu)。譬如 CPU 密集的作業(yè)或許需要更多的核，但不需要太多內(nèi)存，固定結(jié)構(gòu)的 container 會導致內(nèi)存被浪費。
與容器管理基礎(chǔ)設(shè)施的交互比較笨拙，需要兩個步驟來啟動 Flink 作業(yè): 1.啟動 Flink 守護進程；2.提交作業(yè)。如果作業(yè)被容器化并且將作業(yè)部署作為容器部署的一部分，那么將不再需要步驟2。
On-Yarn 模式下，作業(yè)管理頁面會在作業(yè)完成后消失不可訪問。
Flink 推薦 per job clusters 的部署方式，但是又支持可以在一個集群上運行多個作業(yè)的 session 模式，令人疑惑。
在 Flink 版本 1.5 中引入了 Dispatcher，Dispatcher 是在新設(shè)計里引入的一個新概念。Dispatcher 會從 Client 端接受作業(yè)提交請求并代表它在集群管理器上啟動作業(yè)。
