1.Flink架構及特性分析

Flink是個相當早的專案，開始於2008年，但只在最近才得到注意。Flink是原生的流處理系統，提供high level的API。Flink也提供 API來像Spark一樣進行批處理，但兩者處理的基礎是完全不同的。Flink把批處理當作流處理中的一種特殊情況。在Flink中，所有 的資料都看作流，是一種很好的抽象，因為這更接近於現實世界。

在國外一些社群，有很多人將大資料的計算引擎分成了 4 代，當然，也有很多人不會認同。首先第一代的計算引擎，無疑就是 Hadoop 承載的 MapReduce。這裡大家應該都不會對 MapReduce 陌生，它將計算分為兩個階段，分別為 Map 和 Reduce。對於上層應用來說，就不得不想方設法去拆分演算法，甚至於不得不在上層應用實現多個 Job 的串聯，以完成一個完整的演算法，例如迭代計算。由於這樣的弊端，催生了支援 DAG 框架的產生。因此，支援 DAG 的框架被劃分為第二代計算引擎。如 Tez 以及更上層的 Oozie。這裡我們不去細究各種 DAG 實現之間的區別，不過對於當時的 Tez 和 Oozie 來說，大多還是批處理的任務。接下來就是以 Spark 為代表的第三代的計算引擎。第三代計算引擎的特點主要是 Job 內部的 DAG 支援（不跨越Job），以及強調的實時計算。在這裡，很多人也會認為第三代計算引擎也能夠很好的執行批處理的 Job。隨著第三代計算引擎的出現，促進了上層應用快速發展，例如各種迭代計算的效能以及對流計算和 SQL 等的支援。Flink 的誕生就被歸在了第四代。這應該主要表現在 Flink 對流計算的支援，以及更一步的實時性上面。當然Flink 也可以支援 Batch 的任務，以及 DAG 的運算。

1.1 基本架構

下面我們介紹下Flink的基本架構, 有三種部署模式，分別是 Local、Standalone Cluster 和 Yarn Cluster。Flink系統的Yarn Cluster架構與Spark類似，是一個基於Master-Slave風格的架構。

當 Flink 叢集啟動後，首先會啟動一個 JobManger 和一個或多個的 TaskManager。由 Client 提交任務給 JobManager， JobManager 再排程任務到各個 TaskManager 去執行，然後 TaskManager 將心跳和統計資訊彙報給 JobManager。 TaskManager 之間以流的形式進行資料的傳輸。上述三者均為獨立的 JVM 程序。

Client 為提交 Job 的客戶端，可以是執行在任何機器上（與 JobManager 環境連通即可）。提交 Job 後，Client 可以結束程序 （Streaming的任務），也可以不結束並等待結果返回。

JobManager 主要負責排程 Job 並協調 Task 做 checkpoint，職責上很像 Storm 的 Nimbus。從 Client 處接收到 Job 和 JAR 包 等資源後，會生成優化後的執行計劃，並以 Task 的單元排程到各個 TaskManager 去執行。

TaskManager 在啟動的時候就設定好了槽位數（Slot），每個 slot 能啟動一個 Task，Task 為執行緒。從 JobManager 處接收需要 部署的 Task，部署啟動後，與自己的上游建立 Netty 連線，接收資料並處理。

JobManager

JobManager是Flink系統的協調者，它負責接收Flink Job，排程組成Job的多個Task的執行。同時，JobManager還負責收集Job 的狀態資訊，並管理Flink叢集中從節點TaskManager。JobManager所負責的各項管理功能，它接收到並處理的事件主要包括：

RegisterTaskManager

在Flink叢集啟動的時候，TaskManager會向JobManager註冊，如果註冊成功，則JobManager會向TaskManager回覆訊息 AcknowledgeRegistration。

SubmitJob

Flink程式內部通過Client向JobManager提交Flink Job，其中在訊息SubmitJob中以JobGraph形式描述了Job的基本資訊。

CancelJob

請求取消一個Flink Job的執行，CancelJob訊息中包含了Job的ID，如果成功則返回訊息CancellationSuccess，失敗則返回訊息 CancellationFailure。

UpdateTaskExecutionState

TaskManager會向JobManager請求更新ExecutionGraph中的ExecutionVertex的狀態資訊，更新成功則返回true。

RequestNextInputSplit

執行在TaskManager上面的Task，請求獲取下一個要處理的輸入Split，成功則返回NextInputSplit。

JobStatusChanged

ExecutionGraph向JobManager傳送該訊息，用來表示Flink Job的狀態發生的變化，例如：RUNNING、CANCELING、 FINISHED等。

TaskManager

TaskManager也是一個Actor，它是實際負責執行計算的Worker，在其上執行Flink Job的一組Task。每個TaskManager負責管理 其所在節點上的資源資訊，如記憶體、磁碟、網路，在啟動的時候將資源的狀態向JobManager彙報。TaskManager端可以分成兩個 階段：

註冊階段

TaskManager會向JobManager註冊，傳送RegisterTaskManager訊息，等待JobManager返回AcknowledgeRegistration，然 後TaskManager就可以進行初始化過程。

可操作階段

該階段TaskManager可以接收並處理與Task有關的訊息，如SubmitTask、CancelTask、FailTask。如果TaskManager無法連線 到JobManager，這是TaskManager就失去了與JobManager的聯絡，會自動進入“註冊階段”，只有完成註冊才能繼續處理Task 相關的訊息。

Client　　

當用戶提交一個Flink程式時，會首先建立一個Client，該Client首先會對使用者提交的Flink程式進行預處理，並提交到Flink叢集中處 理，所以Client需要從使用者提交的Flink程式配置中獲取JobManager的地址，並建立到JobManager的連線，將Flink Job提交給 JobManager。Client會將使用者提交的Flink程式組裝一個JobGraph， 並且是以JobGraph的形式提交的。一個JobGraph是一個 Flink Dataflow，它由多個JobVertex組成的DAG。其中，一個JobGraph包含了一個Flink程式的如下資訊：JobID、Job名稱、配 置資訊、一組JobVertex等。

1.2 基於Yarn層面的架構

基於yarn層面的架構類似spark on yarn模式，都是由Client提交App到RM上面去執行，然後RM分配第一個container去執行 AM，然後由AM去負責資源的監督和管理。需要說明的是，Flink的yarn模式更加類似spark on yarn的cluster模式，在cluster模式 中，dirver將作為AM中的一個執行緒去執行，在Flink on yarn模式也是會將JobManager啟動在container裡面，去做個driver類似 的task排程和分配，YARN AM與Flink JobManager在同一個Container中，這樣AM可以知道Flink JobManager的地址，從而 AM可以申請Container去啟動Flink TaskManager。待Flink成功執行在YARN叢集上，Flink YARN Client就可以提交Flink Job到 Flink JobManager，並進行後續的對映、排程和計算處理。