聊聊 Flink：Flink 的運行時架構

一、運行時架構

上一篇我們可以看到 Flink 的核心組件的 Deploy 層，該層主要涉及了 Flink 的部署模式，Flink 支持多種部署模式：本地、集羣（Standalone/YARN）、雲（GCE/EC2）。

• Local（本地）：單機模式，一般本地開發調試使用，像我們程序寫的 WordCountStream 那個例子，直接運行 main 方法啓動。

• Cluster（集羣）

• Standalone（獨立模式）：Flink 自帶集羣，自己管理資源調度，生產環境也會有所應用。

• YARN（YARN 模式）：計算資源統一由 Hadoop YARN 管理，生產環境應用較多。

• Cloud（雲端）：AliCloud Realtime Compute、Amazon EMR、Huawei Cloud Stream Service 等。

我們這裏主要來介紹 Cluster 集羣的兩種模式 Standalone、YARN。

二、YARN 集羣架構

在講解 Flink 集羣架構之前，我們先了解一下 YARN 集羣架構，我覺得是很有必要的。YARN 集羣總體上是經典的主／從 (Master/Slave) 架構，主要由 ResourceManager、NodeManager、ApplicationMaster 和 Container 等幾個組件構成。

2.1 ResourceManager

以後臺進程的形式運行，負責對集羣資源進行統一管理和任務調度。ResourceManager 的主要職責如下：

• 接收來自客戶端的請求。

• 啓動和管理各個應用程序的 ApplicationMaster。

• 接收來自 ApplicationMaster 的資源申請，併爲其分配 Container。

• 管理 NodeManager，接收來自 NodeManager 的資源和節點健康情況彙報。

2.2 NodeManager

集羣中每個節點上的資源和任務管理器，以後臺進程的形式運行。它會定時向 ResourceManager 彙報本節點上的資源（內存、CPU）使用情況和各個 Container 的運行狀態，同時會接收並處理來自 ApplicationMaster 的 Container 啓動／停止等請求。NodeManager 不會監視任務，它僅監視 Container 中的資源使用情況，例如。如果一個 Container 消耗的內存比最初分配的更多，就會結束該 Container。

2.3 Task

應用程序具體執行的任務。一個應用程序可能有多個任務，例如一個 MapReduce 程序可以有多個 Map 任務和多個 Reduce 任務。

2.4 Container

YARN 中資源分配的基本單位，封裝了 CPU 和內存資源的一個容器，相當於一個 Task 運行環境的抽象。從實現上看，Container 是一個 Java 抽象類，定義了資源信息。應用程序的 Task 將會被髮布到 Container 中運行，從而限定了 Task 使用的資源量。

一個應用程序所需的 Container 分爲兩類：運行 ApplicationMaster 的 Container 和運行各類 Task 的 Container。前者是由 ResourceManager 向內部的資源調度器申請和啓動的，後者是由 ApplicationMaster 向 ResourceManager 申請的，並由 ApplicationMaster 請求 NodeManager 進行啓動。

我們可以將 Container 類比成數據庫連接池中的連接，需要的時候進行申請，使用完畢後進行釋放，而不需要每次獨自創建。

2.5 ApplicationMaster

ApplicationMaster 可在 Container 內運行任何類型的 Task。例如，MapReduce ApplicationMaster 請求一個容器來啓動 Map Task 或 Reduce Task。也可以實現一個自定義的 ApplicationMaster 來運行特定的 Task，以便任何分佈式框架都可以受 YARN 支持，只要實現了相應的 ApplicationMaster 即可。

我們可以這樣認爲：ResourceManager 管理整個集羣，NodeManager 管理集羣中的單個節點，ApplicationMaster 管理單個應用程序（集羣中可能同時有多個應用程序在運行，每個應用程序都有各自的 ApplicationMaster）。

YARN 集羣中應用程序的執行流程如下圖所示：

• 客戶端提交應用程序（可以是 MapReduce 程序、Spark 程序等）到 ResourceManager。

• ResourceManager 分配用於運行 ApplicationMaster 的 Container，然後與 NodeManager 通信，要求它在該 Container 中啓動 ApplicationMaster。ApplicationMaster 啓動後，它將負責此應用程序的整個生命週期。

• ApplicationMaster 向 ResourceManager 註冊（註冊後可以通過 ResourceManager 查看應用程序的運行狀態）並請求運行應用程序各個 Task 所需的 Container（資源請求是對一些 Container 的請求）。如果符合條件，ResourceManager 會分配給 ApplicationMaster 所需的 Container（表達爲 Container ID 和主機名）。

• ApplicationMaster 請求 NodeManager 使用這些 Container 來運行應用程序的相應 Task（即將 Task 發佈到指定的 Container 中運行）。

此外，各個運行中的 Task 會通過 RPC 協議向 ApplicationMaster 彙報自己的狀態和進度，這樣一旦某個 Task 運行失敗，ApplicationMaster 就可以對其重新啓動。當應用程序運行完成時，ApplicationMaster 會向 ResourceManager 申請註銷自己。

三、Flink Standalone 模式

Flink Standalone 模式爲經典的主從 (Master/Slave) 架構，資源調度是 Flink 自己實現的。集羣啓動後，主節點上會啓動一個 JobManager 進程，類似 YARN 集羣的 ResourceManager，因此主節點也稱爲 JobManager 節點；各個從節點上會啓動一個 TaskManager 進程，類似 YARN 集羣的 NodeManager，因此從節點也稱爲 TaskManager 節點。從 Flink 1.6 版本開始，將主節點上的進程名稱改爲了 StandaloneSessionClusterEntrypoint，從節點的進程名稱改爲了 TaskManagerRunner，在這裏爲了方便使用，仍然沿用之前版本的稱呼，即 JobManager 和 TaskManager。

Client 接收到 Flink 應用程序後，將作業提交給 JobManager。JobManager 要做的第一件事就是分配 Task（任務）所需的資源。完成資源分配後，Task 將被 JobManager 提交給相應的 TaskManager，TaskManager 會啓動線程開始執行。在執行過程中，TaskManager 會持續向 JobManager 彙報狀態信息，例如開始執行、進行中或完成等狀態。作業執行完成後，結果將通過 JobManager 發送給 Client。

Flink 所有組件之間的通信使用的是 Akka 框架，組件之間的數據交互使用的是 Netty 框架。

Client 不是運行時和程序執行的一部分，而是用於準備數據流並將其發送給 JobManager。之後，客戶端可以斷開連接（分離模式），或保持連接來接收進程報告（附加模式）。客戶端可以作爲觸發執行 Java/Scala 程序的一部分運行，也可以在命令行進程./bin/flink run … 中運行。

可以通過多種方式啓動 JobManager 和 TaskManager：直接在機器上作爲 standalone 集羣啓動、在容器中啓動、或者通過 YARN 等資源框架管理並啓動。TaskManager 連接到 JobManagers，宣佈自己可用，並被分配工作。

3.1 JobManager

JobManager 具有許多與協調 Flink 應用程序的分佈式執行有關的職責：它決定何時調度下一個 task（或一組 task）、對完成的 task 或執行失敗做出反應、協調 checkpoint、並且協調從失敗中恢復等等。這個進程由三個不同的組件組成：

• ResourceManager

  ResourceManager 負責 Flink 集羣中的資源提供、回收、分配 - 它管理 task slots，這是 Flink 集羣中資源調度的單位（請參考 TaskManagers）。Flink 爲不同的環境和資源提供者（例如 YARN、Kubernetes 和 standalone 部署）實現了對應的 ResourceManager。在 standalone 設置中，ResourceManager 只能分配可用 TaskManager 的 slots，而不能自行啓動新的 TaskManager。

• Dispatcher

  Dispatcher 提供了一個 REST 接口，用來提交 Flink 應用程序執行，併爲每個提交的作業啓動一個新的 JobMaster。它還運行 Flink WebUI 用來提供作業執行信息。

• JobMaster

  JobMaster 負責管理單個 JobGraph 的執行。Flink 集羣中可以同時運行多個作業，每個作業都有自己的 JobMaster。

始終至少有一個 JobManager。高可用（HA）設置中可能有多個 JobManager，其中一個始終是 leader，其他的則是 standby。

3.2 TaskManager

TaskManager 是 Flink 集羣的工作進程。Task 被調度到 TaskManager 上執行。TaskManager 相互通信，只爲在後續的 Task 之間交換數據。

TaskManager 的主要作用如下：

• 接收 JobManager 分配的任務，負責具體的任務執行。

• TaskManager 會在同一個 JVM 進程內以多線程的方式執行任務。· 負責對應任務在每個節點上的資源申請，管理任務的啓動、停止、銷燬、異常恢復等生命週期。

• 負責對數據進行緩存。TaskManager 之間採用數據流的形式進行數據交互。

3.3 Tasks 和算子鏈

對於分佈式執行，Flink 將算子的 subtasks _鏈接_成 tasks。每個 task 由一個線程執行。將算子鏈接成 task 是個有用的優化：它減少線程間切換、緩衝的開銷，並且減少延遲的同時增加整體吞吐量。鏈行爲是可以配置的。

下圖中樣例數據流用 5 個 subtask 執行，因此有 5 個並行線程。

3.4 Task Slots 和資源

每個 worker（TaskManager）都是一個 JVM 進程，可以在單獨的線程中執行一個或多個 subtask。爲了控制一個 TaskManager 中接受多少個 task，就有了所謂的 task slots（至少一個）。

每個 task slot 代表 TaskManager 中資源的固定子集。例如，具有 3 個 slot 的 TaskManager，會將其託管內存 1/3 用於每個 slot。分配資源意味着 subtask 不會與其他作業的 subtask 競爭託管內存，而是具有一定數量的保留託管內存。注意此處沒有 CPU 隔離；當前 slot 僅分離 task 的託管內存。

通過調整 task slot 的數量，用戶可以定義 subtask 如何互相隔離。每個 TaskManager 有一個 slot，這意味着每個 task 組都在單獨的 JVM 中運行（例如，可以在單獨的容器中啓動）。具有多個 slot 意味着更多 subtask 共享同一 JVM。同一 JVM 中的 task 共享 TCP 連接（通過多路複用）和心跳信息。它們還可以共享數據集和數據結構，從而減少了每個 task 的開銷。

默認情況下，Flink 允許 subtask 共享 slot，即便它們是不同的 task 的 subtask，只要是來自於同一作業即可。結果就是一個 slot 可以持有整個作業管道。允許 _slot 共享_有兩個主要優點：

• Flink 集羣所需的 task slot 和作業中使用的最大並行度恰好一樣。無需計算程序總共包含多少個 task（具有不同並行度）。

• 容易獲得更好的資源利用。如果沒有 slot 共享，非密集 subtask（source/map()）將阻塞和密集型 subtask（window） 一樣多的資源。通過 slot 共享，我們示例中的基本並行度從 2 增加到 6，可以充分利用分配的資源，同時確保繁重的 subtask 在 TaskManager 之間公平分配。

四、Flink On YARN 模式

Flink On YARN 模式遵循 YARN 的官方規範，YARN 只負責資源的管理和調度，運行哪種應用程序由用戶自己實現，因此可能在 YARN 上同時運行 MapReduce 程序、Spark 程序、Flink 程序等。YARN 很好地對每一個程序實現了資源的隔離，這使得 Spark、MapReduce、Flink 等可以運行於同一個集羣中，共享集羣存儲資源與計算資源。Flink On YARN 模式的運行架構如下圖所示。

• 當啓動一個 Client（客戶端）會話時，Client 首先會上傳 Flink 應用程序 JAR 包和配置文件到 HDFS。

• Client 向 ResourceManager 申請用於運行 ApplicationMaster 的 Container。

• ResourceManager 分配用於運行 ApplicationMaster 的 Container，然後與 NodeManager 通信，要求它在該 Container 中啓動 ApplicationMaster（ApplicationMaster 與 Flink JobManager 運行於同一 Container 中，這樣 ApplicationMaster 就能知道 Flink JobManager 的地址）。ApplicationMaster 啓動後，它將負責此應用程序的整個生命週期。另外，ApplicationMaster 還提供了 Flink 的 WebUI 服務。

• ApplicationMaster 向 ResourceManager 註冊（註冊後可以通過 ResourceManager 查看應用程序的運行狀態）並請求運行 Flink TaskManager 所需的 Container（資源請求是對一些 Container 的請求）。如果符合條件，ResourceManager 會分配給 ApplicationMaster 所需的 Container（表達爲 Container ID 和主機名）。ApplicationMaster 請求 NodeManager 使用這些 Container 來運行 Flink TaskManager。各個 NodeManager 從 HDFS 中下載 Flink JAR 包和配置文件。至此，Flink 相關任務就可以運行了。

此外，各個運行中的 Flink TaskManager 會通過 RPC 協議向 ApplicationMaster 彙報自己的狀態和進度。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/CsGdmM4--MCy-eiu86SIOw