2.2 Apache Flink 運行架構

1. 任務提交流程

Flink任務提交後，Client向HDFS上傳Flink的Jar包和配置，以後向Yarn ResourceManager提交任務，ResourceManager分配Container資源並通知對應的NodeManager啓動ApplicationMaster，ApplicationMaster啓動後加載Flink的Jar包和配置構建環境，而後啓動JobManager，以後ApplicationMaster向ResourceManager申請資源啓動TaskManager，ResourceManager分配Container資源後，由ApplicationMaster通知資源所在節點的NodeManager啓動TaskManager，NodeManager加載Flink的Jar包和配置構建環境並啓動TaskManager，TaskManager啓動後向JobManager發送心跳包，並等待JobManager向其分配任務。

2. 任務調度原理

客戶端不是運行時和程序執行的一部分，但它用於準備併發送dataflow給Master，而後，客戶端斷開鏈接或者維持鏈接以等待接收計算結果，客戶端能夠以兩種方式運行：要麼做爲Java/Scala程序的一部分被程序觸發執行，要麼以命令行./bin/flink run的方式執行。

3. Worker與Slots

每個worker(TaskManager)是一個JVM進程，它可能會在獨立的線程上執行一個或多個subtask。爲了控制一個worker能接收多少個task，worker經過task slot來進行控制（一個worker至少有一個task slot）。·

每一個task slot表示TaskManager擁有資源的一個固定大小的子集。假如一個TaskManager有三個slot，那麼它會將其管理的內存分紅三份給各個slot。資源slot化意味着一個subtask將不須要跟來自其餘job的subtask競爭被管理的內存，取而代之的是它將擁有必定數量的內存儲備。須要注意的是，這裏不會涉及到CPU的隔離，slot目前僅僅用來隔離task的受管理的內存。

經過調整task slot的數量，容許用戶定義subtask之間如何互相隔離。若是一個TaskManager一個slot，那將意味着每一個task group運行在獨立的JVM中（該JVM多是經過一個特定的容器啓動的），而一個TaskManager多個slot意味着更多的subtask能夠共享同一個JVM。而在同一個JVM進程中的task將共享TCP鏈接（基於多路複用）和心跳消息。它們也可能共享數據集和數據結構，所以這減小了每一個task的負載。

Task Slot是靜態的概念，是指TaskManager具備的併發執行能力，能夠經過參數taskmanager.numberOfTaskSlots進行配置，而並行度parallelism是動態概念，即TaskManager運行程序時實際使用的併發能力，能夠經過參數parallelism.default進行配置。

也就是說，假設一共有3個TaskManager，每個TaskManager中的分配3個TaskSlot，也就是每一個TaskManager能夠接收3個task，一共9個TaskSlot，若是咱們設置parallelism.default=1，即運行程序默認的並行度爲1，9個TaskSlot只用了1個，有8個空閒，所以，設置合適的並行度才能提升效率。

4. 程序與數據流

Flink程序的基礎構建模塊是 流（streams） 與 轉換（transformations）（須要注意的是，Flink的DataSet API所使用的DataSets其內部也是stream）。一個stream能夠當作一箇中間結果，而一個transformations是以一個或多個stream做爲輸入的某種operation，該operation利用這些stream進行計算從而產生一個或多個result stream。

在運行時，Flink上運行的程序會被映射成streaming dataflows，它包含了streams和transformations operators。每個dataflow以一個或多個sources開始以一個或多個sinks結束。dataflow相似於任意的有向無環圖（DAG），固然特定形式的環能夠經過iteration構建。在大部分狀況下，程序中的transformations跟dataflow中的operator是一一對應的關係，但有時候，一個transformation可能對應多個operator。

5. 並行數據流

Flink程序的執行具備並行、分佈式的特性。在執行過程當中，一個 stream 包含一個或多個 stream partition ，而每個 operator 包含一個或多個 operator subtask，這些operator subtasks在不一樣的線程、不一樣的物理機或不一樣的容器中彼此互不依賴得執行。

一個特定operator的subtask的個數被稱之爲其parallelism(並行度)。一個stream的並行度老是等同於其producing operator的並行度。一個程序中，不一樣的operator可能具備不一樣的並行度。

Stream在operator之間傳輸數據的形式能夠是one-to-one(forwarding)的模式也能夠是redistributing的模式，具體是哪種形式，取決於operator的種類。

One-to-one：stream(好比在source和map operator之間)維護着分區以及元素的順序。那意味着map operator的subtask看到的元素的個數以及順序跟source operator的subtask生產的元素的個數、順序相同，map、fliter、flatMap等算子都是one-to-one的對應關係。

Redistributing：stream(map()跟keyBy/window之間或者keyBy/window跟sink之間)的分區會發生改變。每個operator subtask依據所選擇的transformation發送數據到不一樣的目標subtask。例如，keyBy() 基於hashCode重分區、broadcast和rebalance會隨機從新分區，這些算子都會引發redistribute過程，而redistribute過程就相似於Spark中的shuffle過程。

6. task與operator chains

出於分佈式執行的目的，Flink將operator的subtask連接在一塊兒造成task，每一個task在一個線程中執行。將operators連接成task是很是有效的優化：它能減小線程之間的切換和基於緩存區的數據交換，在減小時延的同時提高吞吐量。連接的行爲能夠在編程API中進行指定。

下面這幅圖，展現了5個subtask以5個並行的線程來執行：