【Flink】深入理解Flink-On-Yarn模式
1. 前言
Flink提供了兩種在yarn上執行的模式，分別為Session-Cluster和Per-Job-Cluster模式，本文分析兩種模式及啟動流程。
下圖展示了Flink-On-Yarn模式下涉及到的相關類圖結構
2. Session-Cluster模式
Session-Cluster模式需要先啟動叢集，然後再提交作業，接著會向yarn申請一塊空間後，資源永遠保持不變。如果資源滿了，下一個作業就無法提交，只能等到yarn中的其中一個作業執行完成後，釋放了資源，下個作業才會正常提交。所有作業共享Dispatcher和ResourceManager；共享資源；適合規模小執行時間短的作業。
2.1. 啟動叢集
執行bin/yarn-session.sh即可預設啟動包含一個TaskManager（記憶體大小為1024MB，包含一個Slot）、一個JobMaster（記憶體大小為1024MB），當然可以通過指定引數控制叢集的資源，如-n指定TaskManager個數，-s指定每個TaskManager中Slot的個數；其他配置項，可參考
下面以bin/yarn-session.sh為例，分析Session-Cluster啟動流程。
2.2. 流程分析
下面分為本地和遠端分析啟動流程，其中本地表示在客戶端的啟動流程，遠端則表示通過Yarn拉起Container的流程；
2.2.1 本地流程
- Session啟動入口為FlinkYarnSessionCli#main
- 根據傳入的引數確定叢集的資源資訊（如多少個TaskManager，Slot等）
- 部署叢集AbstractYarnClusterDescriptor#deploySessionCluster -> AbstractYarnClusterDescriptor#deployInternal
- 進行資源校驗（如記憶體大小、vcore大小、佇列）
- 通過YarnClient建立Application
- 再次校驗資源
- AbstractYarnClusterDescriptor#startAppMaster啟動AppMaster
- 初始化檔案系統（HDFS）
- 將log4j、logback、flink-conf.yaml、jar包上傳至HDFS
- 構造AppMaster的Container（確定Container程序的入口類YarnSessionClusterEntrypoint），構造相應的Env
- YarnClient向Yarn提交Container申請
- 跟蹤ApplicationReport狀態（確定是否啟動成功，可能會由於資源不夠，一直等待）
- 啟動成功後將對應的ip和port寫入flinkConfiguration中
- 建立與將叢集互動的ClusterClient
- 根據flink-conf的HA配置建立對應的服務（如StandaloneHaServices、ZooKeeperHaServices等）
- 建立基於Netty的RestClient；
- 建立/rest_server_lock、/dispatcher_lock節點（以ZK為例）
- 啟動監聽節點的變化（主備切換）
- 通過ClusterClient獲取到appId資訊並寫入本地臨時檔案
經過上述步驟，整個客戶端的啟動流程就結束了，下面分析yarn拉起Session叢集的流程，入口類在申請Container時指定為YarnSessionClusterEntrypoint。
2.2.2 遠端流程
遠端宿主在Container中的叢集入口為YarnSessionClusterEntrypoint#main
ClusterEntrypoint #runClusterEntrypoint -> ClusterEntrypoint#startCluster啟動叢集
- 初始化檔案系統
- 初始化各種Service(如：建立RpcService（AkkaRpcService）、建立HAService、建立並啟動BlobServer、建立HeartbeatServices、建立指標服務並啟動、建立本地儲存ExecutionGraph的Store）
- 建立DispatcherResourceManagerComponentFactory（SessionDispatcherResourceManagerComponentFactory），用於建立DispatcherResourceManagerComponent（用於啟動Dispatcher、ResourceManager、WebMonitorEndpoint）
- 通過DispatcherResourceManagerComponentFactory建立DispatcherResourceManagerComponent
- 建立/dispatcher_lock節點，/resource_manager_lock節點
- 建立DispatcherGateway、ResourceManagerGateway的Retriever（用於建立RpcGateway）
- 建立DispatcherGateway的WebMonitorEndpoint並啟動
- 建立JobManager的指標組
- 建立ResourceManager、Dispatcher並啟動進行ZK選舉
- 返回SessionDispatcherResourceManagerComponent
經過上述步驟就完成了叢集的啟動；
2.3. 啟動任務
當啟動集群后，即可使用./flink run -c mainClass /path/to/user/jar向叢集提交任務。
2.4 流程分析
同樣，下面分為本地和遠端分析啟動流程，其中本地表示在客戶端提交任務流程，遠端則表示叢集收到任務後的處理流程。
2.4.1 本地流程
- 程式入口為CliFrontend#main
- 解析處理引數
- 根據使用者jar、main、程式引數、savepoint資訊生成PackagedProgram
- 獲取session叢集資訊
- 執行使用者程式
- 設定ClassLoader
- 設定Context
- 執行使用者程式main方法（當執行使用者業務邏輯程式碼時，會解析出StreamGraph然後通過ClusterClient#run來提交任務），其流程如下：
- 獲取任務的JobGraph
- 通過RestClusterClient#submitJob提交任務
- 建立本地臨時檔案儲存JobGraph
- 通過RestClusterClient向叢集的rest介面提交任務
- 處理請求響應結果
- 重置Context
- 重置ClassLoader
經過上述步驟，客戶端提交任務過程就完成了，主要就是通過RestClusterClient將使用者程式的JobGraph通過Rest介面提交至叢集中。
2.4.2 遠端流程
遠端響應任務提交請求的是RestServerEndpoint，其包含了多個Handler，其中JobSubmitHandler用來處理任務提交的請求；
- 處理請求入口：JobSubmitHandler#handleRequest
- 進行相關校驗
- 從檔案中讀取出JobGraph
- 通過BlobClient將jar及JobGraph檔案上傳至BlobServer中
- 通過Dispatcher#submitJob提交JobGraph
- 通過Dispatcher#runJob執行任務
- 建立JobManagerRunner（處理leader選舉）
- 建立JobMaster（實際執行任務入口，包含在JobManagerRunner）
- 啟動JobManagerRunner（會進行leader選舉，ZK目錄為leader/${jobId}/job_manager_lock）
- 當為主時會呼叫JobManagerRunner#grantLeadership方法
- 啟動JobMaster
- 將任務執行狀態資訊寫入ZK（/${AppID}/running_job_registry/${jobId}）
- 啟動JobMaster的Endpoint
- 開始排程任務JobMaster#startJobExecution
接下來就進行任務具體排程（構造ExecutionGraph、申請Slot等）流程，本篇文章不再展開介紹。
3. Per-Job-Cluster模式
一個任務會對應一個Job，每提交一個作業會根據自身的情況，都會單獨向yarn申請資源，直到作業執行完成，一個作業的失敗與否並不會影響下一個作業的正常提交和執行。獨享Dispatcher和ResourceManager，按需接受資源申請；適合規模大長時間執行的作業。
3.1 啟動任務
啟動Per-Job-Cluster任務，可通過./bin/flink run -m yarn-cluster -d -c mainClass /path/to/user/jar命令使用分離模式啟動一個叢集，即單任務單叢集；
3.2. 流程分析
與Session-Cluster類似，我們對Per-Job-Cluster模式也分為本地和遠端。
3.2.1 本地流程
- 與Session-Cluster模式類似，入口也為CliFrontend#main
- 解析處理引數
- 根據使用者jar、main、程式引數、savepoint資訊生成PackagedProgram
- 根據PackagedProgram建立JobGraph（對於非分離模式還是和Session模式一樣，模式Session-Cluster）
- 獲取叢集資源資訊
- 部署叢集YarnClusterDesriptor#deployJobCluster -> AbstractYarnClusterDescriptor#deployInternal；後面流程與Session-Cluster類似，值得注意的是在AbstractYarnClusterDescriptor#startAppMaster中與Session-Cluster有一個顯著不同的就是其會將任務的JobGraph上傳至Hdfs供後續服務端使用
經過上述步驟，客戶端提交任務過程就完成了，主要涉及到檔案（JobGraph和jar包）的上傳。
3.2.2 遠端流程
- 遠端宿主在Container中的叢集入口為YarnJobClusterEntrypoint#main
- ClusterEntrypoint#runClusterEntrypoint -> ClusterEntrypoint#startCluster啟動叢集
- 建立JobDispatcherResourceManagerComponentFactory（用於建立JobDispatcherResourceManagerComponent）
- 建立ResourceManager（YarnResourceManager）、Dispatcher（MiniDispatcher），其中在建立MiniDispatcher時會從之前的JobGraph檔案中讀取出JobGraph，並啟動進行ZK選舉
- 當為主時會呼叫Dispatcher#grantLeadership方法
- Dispatcher#recoverJobs恢復任務，獲取JobGraph
- Dispatcher#tryAcceptLeadershipAndRunJobs確認獲取主並開始執行任務
- Dispatcher#runJob開始執行任務（建立JobManagerRunner並啟動進行ZK選舉），後續流程與Session-Cluster相同，不再贅述
4. 總結
Flink提供在Yarn上兩種執行模式：Session-Cluster和Per-Job-Cluster，其中Session-Cluster的資源在啟動叢集時就定義完成，後續所有作業的提交都共享該資源，作業可能會互相影響，因此比較適合小規模短時間執行的作業，對於Per-Job-Cluster而言，所有作業的提交都是單獨的叢集，作業之間的執行不受影響（可能會共享CPU計算資源），因此比較適合大規模長時間執行的作業