1.3.6　Flink程序執(zhí)行模式

Flink可以通過以下3種方式來執(zhí)行應用程序：①以Application模式；②以Per-Job模式；③以Session模式。這3種方式如圖1-21所示。

以上3種模式的區(qū)別在于：

（1）集群生命周期和資源隔離保證。

（2）應用程序的main（）方法是在客戶端上執(zhí)行還是在集群上執(zhí)行。

1．Per-Job模式

為了提供更好的資源隔離保證，Per-Job模式使用可用的資源提供者框架（例如YARN、Kubernetes）為每個提交的作業(yè)啟動一個集群。該集群僅對該作業(yè)可用。當作業(yè)完成時，將關閉集群，并清除所有滯留的資源（文件等）。

（1）集群生命周期：在Flink Job Cluster中，可用的集群管理器（如YARN）用于為每個提交的作業(yè)啟動集群，該集群僅對該作業(yè)可用。在這里，客戶端首先從集群管理器請求資源以啟動JobManager，并將作業(yè)提交給運行在此進程中的Dispatcher，然后根據(jù)作業(yè)的資源需求惰性地分配TaskManager。一旦作業(yè)完成，F(xiàn)link作業(yè)集群就會被拆除。集群的生命周期與作業(yè)的生命周期綁定在一起。

（2）資源隔離：更好的隔離保證，因為資源不會跨作業(yè)共享。JobManager中的致命錯誤只會影響Flink Job Cluster中運行的一個作業(yè)。此外，它將負載分散到多個JobManager，因為每個作業(yè)有一個JobManager。

由于ResourceManager需要申請并等待外部資源管理組件啟動TaskManager進程并分配資源，所以Flink Job Clusters更適合長時間運行、對穩(wěn)定性要求高、對啟動時間不敏感的大型任務。Per-Job資源分配模型是許多生產環(huán)境的首選模式（注：Kubernetes和Standalone集群不支持此模式）。

采用這種模式的有Flink On YARN，JobManager不會預先啟動，此時Client將首先向資源管理系統(tǒng)（如YARN、K8S）申請資源來啟動JobManager，然后向JobManager中的Dispatcher提交作業(yè)。

例如，可以在YARN上使用Per-Job模式來部署并運行Flink自帶的示例程序，命令如下：

這將在YARN上啟動一個Flink集群，然后在本地運行提供的應用程序JAR，最后將JobGraph提交給YARN上的JobManager。如果傳遞--detached參數(shù)，則客戶端將在提交被接受后停止。一旦作業(yè)停止，YARN集群就會停止Flink集群。

一旦部署了Per-Job Cluster，就可以與它進行交互了，以執(zhí)行取消或獲取保存點等操作。常用命令如下：

需要注意，取消Per-Job Cluster上的作業(yè)將停止集群。

2．Session模式

Session模式假設有一個已經運行的集群（稱為Flink Session Cluster），并使用該集群的資源來執(zhí)行任何提交的應用程序。在Session Mode中，集群的生命周期獨立于集群上運行的任何作業(yè)的生命周期，并且資源在所有作業(yè)之間共享。

（1）集群生命周期：在Session模式中，客戶端連接到一個預先存在的、長時間運行的集群，該集群可以接受多個作業(yè)提交，即使在所有作業(yè)完成后，集群（和JobManager）仍將繼續(xù)運行，直到手動停止會話，因此，F(xiàn)link Session Cluster的生命周期不與任何Flink Job的生命周期綁定。

（2）資源隔離：TaskManager槽在作業(yè)提交時由ResourceManager分配，作業(yè)完成后釋放。因為所有作業(yè)都共享同一個集群，所以在提交作業(yè)階段存在一些對集群資源的競爭，例如網絡帶寬。這種共享設置的一個限制是，如果一個TaskManager崩潰了，則所有在這個TaskManager上運行的任務都會失敗；以類似的方式，如果JobManager上發(fā)生一些致命錯誤，則它將影響集群中運行的所有作業(yè)。除了對導致失敗的作業(yè)產生負面影響外，這還意味著一個潛在的大規(guī)模恢復過程，所有重新啟動的作業(yè)都同時訪問文件系統(tǒng)，并使其他服務無法使用它。此外，讓一個集群運行多個作業(yè)意味著JobManager需要承擔更多的負載，因為需要JobManager負責對集群中的所有作業(yè)進行記賬。

使用預先存在的集群可以節(jié)省大量申請資源和啟動TaskManager的時間。當作業(yè)的執(zhí)行時間非常短，而高啟動時間會對端到端用戶體驗產生負面影響時，這一點非常重要，就像短查詢的交互分析那樣，作業(yè)可以使用現(xiàn)有資源快速執(zhí)行計算。

采用這種模式的有Standalone，JobManager會預先啟動，此時Client直接與Dispatcher建立連接并提交作業(yè)即可。

例如，使用Session Mode部署Flink自帶的WordCount程序，命令如下：

3．Application模式

在Per-Job模式和Session模式中，應用程序的main（）方法都在客戶端執(zhí)行。這個過程包括在本地下載應用程序的依賴項，執(zhí)行main（）來提取Flink的運行時可以理解的應用程序的表示（JobGraph），并將依賴項和JobGraph發(fā)送到集群。這使客戶端成為一個沉重的資源消耗者，因為它可能需要大量的網絡帶寬來下載依賴項并將二進制文件發(fā)送到集群，并需要CPU周期性地執(zhí)行main（）。當客戶端在用戶之間共享時，這個問題會更加明顯。

在此觀察的基礎上，Application模式為每個提交的應用程序創(chuàng)建一個集群，并且應用程序的main（）方法在JobManager上執(zhí)行。在JobManager上執(zhí)行main（）可以節(jié)省所需的CPU周期，還可以節(jié)省本地下載依賴項所需的帶寬。此外，為了下載集群中應用程序的依賴關系，它允許更均勻地分散網絡負載，因為每個應用程序有一個JobManager。

（1）集群生命周期：一個Flink Application Cluster是一個專用的Flink集群，它只執(zhí)行來自一個Flink應用程序的作業(yè)，其中main（）方法運行在集群而不是客戶機上。作業(yè)提交是one-step過程：不首先需要啟動一個Flink集群，然后將作業(yè)提交到現(xiàn)有的集群會話；相反，可以將應用程序邏輯和依賴打包到一個可執(zhí)行的作業(yè)JAR中，并且集群入口點（ApplicationClusterEntryPoint）負責調用main（）方法來提取JobGraph。這允許用戶像在Kubernetes上部署其他應用程序一樣部署Flink應用程序，因此，F(xiàn)link Application Cluster集群的生命周期與Flink應用程序的生命周期綁定在一起。

（2）資源隔離：在Flink Application Cluster中，ResourceManager和Dispatcher被限定為單個Flink應用程序，這比Flink Session Cluster提供了更好的關注點分離。

與Per-Job模式相比，Application模式允許提交由多個作業(yè)組成的應用程序。作業(yè)執(zhí)行的順序不受部署模式的影響，而是受用于啟動作業(yè)的調用的影響。使用execute（）是阻塞的，它會建立一個順序，并導致“下一個”作業(yè)的執(zhí)行被推遲到“這個”作業(yè)完成。使用executeAsync（）方法，它是非阻塞的，將導致“下一個”任務在“這個”任務完成之前開始。

Application模式相當于為每個應用程序創(chuàng)建一個會話集群，并在集群上執(zhí)行應用程序的main（）方法。

在這種部署模式下，應用程序JAR文件需要在類路徑中可用。最簡單的方法就是把JAR放到lib/文件夾中。例如，使用Application模式部署Flink自帶的TopSpeedWindowing程序，命令如下：

該腳本還支持停止服務。如果想停止多個實例，則可多次調用它們，或者使用stop-all，命令如下：