Distributed Runtime Environment（分布式運行環境）

Tasks and Operator Chains

在分布式執行情況下，Flink將operator subtasks 鏈接到一起，形成任務（task）。每個任務（subtask）由一個線程執行。將operator subtasks鏈接到任務中是一個好處：它減少了線程到線程的切換和緩沖的開銷，並在減少延遲的同時提高了總體吞吐量。鏈接行為是可以進行配置的;有關詳細信息，請參閱此文檔。

下圖中有5個子任務，因此就有5個並行的線程。對圖中內容解釋下：source和map是兩個operator subtasks，組成一個subtask；keyBy()，window()和apply()三個operator subtasks組成一個subtask。

Job Managers, Task Managers, Clients
Flink的運行過程中有兩個進程：

• JobManagers（master）：協調分布式的執行。他們調度任務，協調檢查點，協調故障恢復等。至少需要有一個Job Manager，有一個是leader。一個高可用的環境，會有多個Job Manager，永遠只有一個是leader，其它則是備用。
• TaskManager（worker）：執行數據流中的子任務，緩沖和交換數據流。至少有一個TaskManager。

JobManagers and TaskManagers運行方式：

• standalone cluster
• YARN
• Mesos

TaskManagers連接到JobManagers，告訴JobManagers自己處於可用狀態，可以用來執行任務。

Client不是Flink運行時和程序執行的一部分，而是用於準備並向JobManager發送數據流。之後，客戶端可以斷開連接，或者保持連接以接收進度報告。Client可以作為觸發執行的Java/Scala程序的一部分運行，也可以在命令行進程中運行比如說：./bin/flink run ....

Task Slots and Resources（任務槽與資源）

每個TaskManager（worker）是一個JVM進程，可以執行一個或者多個子任務在不同的線程中。為了控制一個TaskManager（worker）接受多少任務，一個TaskManager至少有一個任務槽。

每個任務槽(task slot)擁有TaskManager固定的資源。例如，一個TaskManager有三個任務槽(task slot)，那麽每個任務槽(task slot)將會占用TaskManager內存的1/3。任務槽之間的內存資源不存在競爭關系。但是，這裏沒有發生CPU隔離;目前（Flink1.6），槽只分離，任務的托管內存。

通過調整任務槽的數量，用戶可以定義子任務如何相互隔離。每個TaskManager有一個task slot意味著每個任務組在一個單獨的JVM中運行(例如，它可以在一個單獨的container中啟動)。擁有多個task slot意味著更多的子任務共享同一個JVM。相同JVM中的任務共享TCP連接(通過多路復用)和心跳消息。它們還可以共享數據集和數據結構，從而減少每個任務的開銷。

默認情況下，Flink允許子任務（subtasks）共享任務槽（task slots），即使它們是不同任務的子任務，只要它們來自同一個任務。也就是一個槽可以容納整個作業管道。允許這種槽共享有兩個主要好處:

• Flink集群只需要與作業（job）中使用的最高並行度一樣多的任務槽即可。不需要計算程序總共包含多少個任務(每個任務中可能具有不同的並行度)。
• 便於將資源利用最大化。如果沒有槽共享，非密集型source/map()子任務將占據與資源密集型窗口子任務一樣多的資源。通過task slot共享，將下圖示例中的基本並行度從2提高到6，可以充分利用有槽資源，同時確保繁重的子任務在task manager中公平分配。

對圖中內容解釋一下：圖中包含2個進程，也就是兩個TaskManager（兩個JVM進程）；每個TaskManager有三個task slot，如圖現在在task slot共享的情況下，並行度是6。如果不存在task slot共享，那麽6個task slot，有2個task slot執行source()，map()操作，兩個task slot執行keyBy()，window()和apply()操作，sink操作至少需要一個task slot。這就是基本並行度從2提高到6。

APIs中還包括一個資源組機制，可以用來防止不需要的task slot共享。根據經驗，一個好的默認任務槽數是CPU內核數（應用中這樣配置就可以）。對於超線程，每個task slot需要2個或更多的硬件線程上下文。

State Backends

鍵/值索引存儲的確切數據結構取決於所選的狀態後端（State Backends）。一個狀態後端（State Backends）將數據存儲在一個內存hash map中，另一個狀態後端使用RocksDB作為key/value存儲。
除了定義保存狀態的數據結構外，狀態後端（State Backends）還實現了獲取key/value狀態的時間點快照的邏輯，並將該快照存儲為檢查點（checkpoint）的一部分。

Savepoints

程序可以從保存點（Savepoints）恢復執行。保存點允許在不丟失任何狀態的情況下，更新程序和Flink集群。

保存點（Savepoints）是手動觸發的檢查點，它獲取程序的快照並將其寫入狀態後端（ state backend.）。它們依賴於規則的檢查點（checkpoints）機制。在程序執行過程中，在工作節點（works）會定期對程序進行快照，並生成檢查點（checkpoints）。對於恢復操作來說，只需要最後一個完成的檢查點即可；當新的檢查點完成，舊的檢查點可以安全地丟棄。

保存點（Savepoints）類似於這些定期檢查點，只是它們是由用戶觸發的，在新的檢查點完成時，舊的檢查點也不會自動過期。保存點可以通過命令行創建，或者在通過REST API取消作業（job）時創建。

Flink1.6系列之—分布式運行環境