Flink 中的應用部署：當前狀態與新應用模式

Flink 中的應用部署：當前狀態與新應用模式

• 發佈文章

Flink 中的應用部署：當前狀態與新應用模式

同步滾動：開

做爲現代企業的重要工具，流處理和實時分析這類工具逐漸興起，愈來愈多的企業以 Apache Flink 爲核心構建平臺，並將其做爲服務在內部提供。在最新舉辦的 Flink Forward 會議中， Uber、 Netflix 和阿里巴巴等公司的許多相關主題演講進一步說明了這一趨勢。

衆多平臺旨在經過減輕最終用戶的全部運營負擔來簡化內部的 Application （應用）提交。爲了提交 Flink 應用程序，這些平臺一般只公開一個集中式或低並行度端點（例如 Web 前端）用於應用提交，咱們將調用 Deployer（部署器）。

平臺開發人員和維護人員常常提到的障礙之一是，Deployer 多是一個很難配置的大量資源消耗者。若是按照平均負載進行配置，可能會致使 Deployer 服務被部署請求淹沒（在最壞的狀況下，短期內對全部生產應用程序都是如此），而按照最高負載進行規劃的話，又會帶來沒必要要的成本。根據這一觀察結果，Flink 1.11 引入了 Application 模式（應用模式）做爲部署選項，它容許一個輕量級、更可伸縮性的應用提交過程，從而使應用程序部署負載更均勻地分佈在集羣的各個節點上。

爲了理解這個問題以及瞭解 Application 模式如何解決該問題，咱們首先簡要概述 Flink 中應用程序執行的當前狀態，而後再闡述部署模式引入的架構變化以及如何利用它們。

Flink 中的應用程序執行

在 Flink 中執行應用程序主要涉及三個實體：Client（客戶端）、JobManager（做業管理器）和 TaskManager（任務管理器）。Client 負責將應用提交給集羣，JobManager 負責執行期間必要的 bookkeeping，而 TaskManager 則負責實際的計算。更多細節請參考 Flink 的架構 文檔。

當前部署模式

在 1.11 版本中引入 Application 模式以前，Flink 容許用戶在 Session（會話）或 Per-Job 集羣上執行應用程序。二者之間的差別與集羣生命週期和它們提供的資源隔離保證有關。

Session 模式

Session 模式（會話模式）假定集羣已經運行，並使用該集羣的資源來執行任何提交的應用程序。在同一（Session）集羣中執行的應用程序使用相同的資源，並所以相互競爭。這樣作的好處是，你無需爲每一個提交的做業分配整個集羣的資源開銷。可是，若是其中一個做業行爲不正常或者關閉了 TaskManager，那麼在該 TaskManager 上運行的全部做業都將受到故障的影響。除了對致使故障的做業產生負面影響以外，這還意味着潛在的大規模恢復過程，即全部從新啓動的做業同時訪問文件系統，並使其不可用於其餘服務。此外，單個集羣運行多個做業意味着 JobManager 的負載更大，它負責集羣中全部做業的 bookkeeping。這種模式很是適合啓動短做業，例如交互式查詢。

Per-Job 模式

在 Per-Job 模式中，可用的集羣管理器框架（如 YARN 或 Kubernetes）用於爲每一個提交的做業啓動 Flink 集羣，該集羣僅對該做業可用。看成業完成後，集羣將關閉，並清理全部延遲的資源（例如文件）。這種模式容許更好的資源隔離，由於行爲不正常的做業不會影響任何其餘做業。另外，因爲每一個應用程序都有本身的 JobManager，所以它將 bookkeeping 負載分散到多個實體。考慮到前面提到的 Session 模式中的資源隔離問題，對於長時間運行的做業，用戶常常選擇 Per-Job 模式，由於這些做業願意接受必定程度的啓動延遲的增長，以支持彈性。

總之，在 Session 模式中，集羣生命週期獨立於集羣中運行的任何做業，而且集羣中運行的全部做業共享其資源。Per-Job 模式選擇爲每一個提交的做業分配一個集羣，已提供更好的資源隔離保證，由於資源不會在做業之間共享。在這種狀況下，集羣的生命週期與做業的生命週期相關。

Application 提交

Flink 應用程序的執行包括兩個階段：pre-flight，即當用戶的 main() 方法被調用時；runtime，即用戶代碼調用 execute()時當即觸發。main() 方法使用 Flink 的 API（DataStream API、Table API、DataSet API）之一構造用戶程序。當 main() 方法調用 env.execute() 時，用戶定義的管道將被轉換成一種 Flink 運行時能夠理解的形式，稱爲 Job Graph（做業圖），並將其傳遞給集羣。

儘管它們方法有所不一樣，Session 模式和 Per-Job 模式都會在 Client 執行應用程序的 main() 方法，即 pre-flight 階段。

對於已經在本地擁有做業的全部依賴項，而後經過在其機器上運行的 Client 提交其應用程序的單個用戶來講，這一般不是問題。可是，對於經過遠程實體（如 Deployer）提交的狀況下，這個過程包括：

• 本地下載應用程序的依賴項；
• 執行 main() 方法提取 Job Graph；
• 將 Job Graph 及其依賴項發送到集羣以便執行；
• 等待結果。

這使得 Client 消耗了大量的資源，由於它可能須要大量的網絡帶寬來下載依賴項或將二進制文件發送到集羣，而且須要 CPU 週期來執行 main() 方法。隨着愈來愈多的用戶共享同一個 Client，這個問題甚至會變得更加突出。

上圖展現了使用紅色、藍色和綠色表示的三個應用程序的兩種部署模式。每一個矩形都有三個並行項。黑色矩形表示不一樣的進程，分別是 TaskManager、JobManager 和 Deployer。咱們假設在全部狀況下只有一個 Deployer 進程。彩色三角形表示提交進程的負載，而彩色矩形表示 TaskManager 和 JobManager 進程的負載。如圖所示，Per-Job 模式和 Session 模式下的 Deployer 共享相同的負載。它們的不一樣之處在於任務的分配和 JobManager 負載。在 Session 模式下，集羣中的全部做業都有一個 JobManager，而在 Per-Job 模式下，每一個做業都有一個 JobManager。此外，Session 模式下的任務會隨機分配給 TaskManager，而在 Per-Job 模式下，每一個 TaskManager 只能有單個做業任務。

Application 模式

Application 模式創建在上述觀察結果的基礎上，並嘗試將 Per-Job 模式的資源隔離與輕量級且可伸縮的應用提交過程結合起來。爲實現這一點，它爲每一個提交的應用程序建立一個集羣，可是這一次，應用程序的 main() 方法在 JobManager 上執行。

爲每一個應用程序建立一個集羣能夠看做是建立一個只在特定應用程序的做業之間共享的 Session 集羣，並在應用程序結束時關閉。使用這種架構，Application 模式提供與 Per-Job 模式相同的資源隔離和負載平衡保證，但在整個應用程序的粒度上。這是有道理的，由於屬於同一應用程序的工做應該相互關聯，並被視爲一個單元。

在 JobManager 上執行 main() 方法不只能夠節省提取 Job Graph 所需的 CPU 週期，也能夠節省 Client 本地下載依賴項並將 Job Graph 及其依賴項發送到集羣所需的帶寬。此外，因爲每一個應用程序只有一個 JobManager，所以，它能夠更均勻地分散網絡負載。上圖對此進行了說明，其中咱們具備與 「Session 和 Per-Job 部署模式」 部分中相同的場景，可是這一次 Client 負載已經轉移到了每一個應用程序的 JobManager。

注：在 Application 模式下， main() 方法是在集羣上執行的，而不是像在其餘模式中那樣在 Client 上執行。和可能對代碼產生影響，例如，使用 regsiterCachedFile() 在環境中註冊的任何路徑都必須由應用程序的 JobManager 進行訪問。

與 Per-Job 模式相比，Application 模式容許提交由多個做業組成的應用程序。做業執行的順序不受部署模式的影響，而是受用於啓動做業的調用的影響。使用阻塞 execute() 方法創建一個順序，並將致使下一個做業的執行被延遲到「這個」做業完成爲止。相反，一旦提交了當前做業，非阻塞 executeAsync() 方法將當即繼續提交「下一個」做業。

下降網絡需求

如上所述，經過在 JobManager 上執行應用程序的 main() 方法，Application 模式能夠節省之前在提交做業時所需的大量資源。但仍有改進的餘地。

重點關注 YARN，它已經支持全部提到的 here 2，即便 Application 模式已經就緒，Client 仍然須要發送用戶 Jar 到 JobManager。此外，對於每一個應用程序，Client 必須將 「flink-dist」 目錄發送到集羣，該目錄包含框架自己的二進制文件，包括 flink-dist.jar 、 lib/ 和 plugin/ 目錄。這二者可能會在 Client 佔用大量帶寬。此外，在每次提交時發送相同的 flink-dist 二進制文件既是對帶寬的浪費，也是對存儲空間的浪費。只需容許應用程序共享相同的二進制文件便可減小存儲空間的浪費。

在 Flink 1.11 中，咱們引入了醫學選項，容許用戶進行以下操做：

1. 指定一個目錄的遠程路徑，在該目錄中，YARN 能夠找到 Flink 分發二進制文件，以及
2. 指定一個遠程路徑，YARN 能夠在其中找到用戶 Jar。

對於第一步，咱們利用了 YARN 的分佈式緩存，並容許應用程序共享這些二進制文件。所以，若是一個應用程序碰巧在它的 TaskManager 的本地存儲中找到了 Flink 的副本，因爲以前的一個應用程序在同一個 TaskManager 上執行，它甚至都不須要在內部下載它。

注：這兩種優化均可以用於 YARN 上的全部部署模式，而不只僅是 Application 模式。

示例：YARN 上的 Application 模式

有關完整說明，請參閱 Flink 的官方文檔，更具體地說，請參閱引用集羣管理框架的頁面，例如 YARN 或 Kubernetes。接下來我將給出一些關於 YARN 的例子，其中上述全部功能都是可用的。

要以 Application 模式啓動用用程序，可使用：

./bin/flink run-application -t yarn-application ./MyApplication.jar

使用這條命令，全部的配置參數，例如用於引導應用程序狀態的保存點的路徑，或者所需的 JobManager/TaskManager 內存大小，均可以經過它們的配置選項（以 -d 做爲前綴）來指定。有關可用配置選項的目錄，請參閱 Flink 的 配置頁面。

例如，指定 JobManager 和 TaskManager 內存大小的命令以下所示：

./bin/flink run-application -t yarn-application \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
./MyApplication.jar

最後，爲了進一步節省提交應用程序 Jar 所需的帶寬，能夠預先將其上傳到 HDFS，並指定指向 ./MyApplication.jar 的遠程路徑，以下所示：

./bin/flink run-application -t yarn-application \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
-Dyarn.provided.lib.dirs="hdfs://myhdfs/remote-flink-dist-dir" \
hdfs://myhdfs/jars/MyApplication.jar

這將使做業提交變得更加輕量級，由於所需的 Flink jar 和應用程序 Jar 將從指定的遠程位置提取，而不是由 Client 發送到集羣。Client 將惟一提供給集羣的是應用程序的配置，其中包括上述提到的全部路徑。

總結

咱們但願本文的討論可以幫助你理解 Flink 提供的各類部署模式之間的差別，而且可以幫助你做出明智的決定，究竟哪種模式適合你本身的設置。

做者介紹：

Kostas Kloudas，Apache Flink PMC Member、Committer，Ververica 軟件工程師。

原文連接：

https://flink.apache.org/news/2020/07/14/application-mode.html