概覽

本篇我們來了解Flink的部署模式和Flink集群的整體架構(gòu)

部署模式

Flink支持如下三種運(yùn)行模式

Flink的部署步驟分為如下2步：

• 部署啟動一個(gè)Flink Cluster,負(fù)責(zé)接收job提交請求和管理job信息;
• 向Flink Cluster提交job; 根據(jù)Flink Cluster可以運(yùn)行的任務(wù)的數(shù)量(1個(gè)或多個(gè))和提交job請求的地點(diǎn)(遠(yuǎn)端或Cluster端)的不同，從而有了不同的運(yùn)行模式。由于pre-job模式已經(jīng)被廢棄了，下面我們主要來學(xué)習(xí)下Application mode和session mode

Application mode

Application mode是Flink Cluster運(yùn)行1個(gè)job，提交任務(wù)的地點(diǎn)為Cluster端。其提交方式如下

./bin/flink run-application -t yarn-application ./examples/streaming/TopSpeedWindowing.jar

其處理流程為，客戶端提交部署請求，服務(wù)端啟動Flink Cluster, 服務(wù)端運(yùn)行Flink Application提交Job到Cluster。下面我們分析下具體實(shí)現(xiàn)細(xì)節(jié)。

客戶端提交請求

通過flink命令提交請求，其運(yùn)行的類為CliFrontend。為支持部署到不同的資源管理平臺，所以有和對應(yīng)資源管理系統(tǒng)交互的類，具體如下：

• CliFrontend:flink命令對應(yīng)的類，發(fā)起提交請求，后面session mode的提交Flink Application也是由該類負(fù)責(zé)
• ClusterClientFactory:集群客戶端工廠類，負(fù)責(zé)生成不同資源管理平臺的客戶端
• ClusterDescriptor:負(fù)責(zé)和對應(yīng)的資源管理平臺交互，申請資源和提交請求
• ClusterEntrypoint:在資源管理平臺運(yùn)行的類，啟動Flink Cluster。 針對不同資源管理平臺的對應(yīng)實(shí)現(xiàn)類如下：

服務(wù)端啟動&提交Application

服務(wù)端啟動對應(yīng)的ClusterEntrypoint，其中會啟動一個(gè)REST Server來接受提交Flink Application，另外有個(gè)Dispatcher負(fù)責(zé)作業(yè)的調(diào)度，其他部分后面我們分析運(yùn)行流程時(shí)再展開介紹。作業(yè)的提交請求是在Dispatcher中的DispatcherBootstrap屬性實(shí)例化的時(shí)候觸發(fā)。 Flink Application運(yùn)行時(shí)，是在StreamExecutionEnvironment.execute()方法來觸發(fā)實(shí)際提交，提交相關(guān)的調(diào)用鏈如下：

這幾個(gè)都是接口類，在Application模式下對應(yīng)的實(shí)現(xiàn)類如下

session mode

session mode是一個(gè)Flink Cluster可以來運(yùn)行多個(gè)Flink job。那這里的提交會分為2個(gè)步驟

// 提交啟動session cluster
// yarn session
./bin/yarn-session.sh --detached
// kubernates session
./bin/kubernetes-session.sh -Dkubernetes.cluster-id=my-first-flink-cluster
// 提交job
./bin/flink run ./examples/streaming/TopSpeedWindowing.jar

• 通過yarn-session.sh (或kubernates-session.sh) 來提交部署Flink Cluster,這塊和前面application mode類似，以yarn模式為例，底層也是調(diào)用了YarnClusterDescriptor來提交相應(yīng)的請求，提交到服務(wù)器的是YarnSessionClusterEntrypoint類。
• 提交Job，這塊是在client端來單獨(dú)提交的，直接提交信息到服務(wù)器的REST Server，根據(jù)提交的目標(biāo)資源管理系統(tǒng)的不同，使用了不同的實(shí)現(xiàn)類

Cluster架構(gòu)

Flink是一個(gè)Master/Worker的架構(gòu)，Master節(jié)點(diǎn)負(fù)責(zé)整個(gè)任務(wù)的管理，Worker節(jié)點(diǎn)負(fù)責(zé)執(zhí)行對應(yīng)的任務(wù)。其整體結(jié)構(gòu)如下：

* JobManager: Master節(jié)點(diǎn)的統(tǒng)稱，目前版本沒有該類，其中有幾個(gè)重點(diǎn)的服務(wù)，如上圖所示，目前的代碼中對應(yīng)的組合了這些服務(wù)的類為：

Dispatcher

ResourceManager

Component。

* Dispatcher: Job調(diào)度器，負(fù)責(zé)接收J(rèn)ob的提交，保存Job和管理JobMaster來執(zhí)行作業(yè)。前面我們提到的提交作業(yè)到Cluster，實(shí)際上是提交給了Dispatcher的。

* ResourceManager: 負(fù)責(zé)和不同的資源調(diào)度系統(tǒng)交互，管理資源申請。

* WebMonitorEndpoint: 負(fù)責(zé)web界面的Rest請求處理

* JobMaster: 負(fù)責(zé)運(yùn)行單個(gè)JobGraph，包括TaskManager的管理，任務(wù)的調(diào)度等。

* TaskManager: 負(fù)責(zé)任務(wù)的執(zhí)行，也沒有TaskManager的類，對應(yīng)的類為TaskExecutor，來執(zhí)行多個(gè)Task

說明：JobManager可能是原來的JobMaster，具體通過Dispatcher.java的如下代碼可以看出，重點(diǎn)在對其具體結(jié)構(gòu)的理解，這個(gè)變化的邏輯我們就不考究了。

 private JobManagerRunner createJobMasterRunner(JobGraph jobGraph) throws Exception

Cluster的啟動流程

上面介紹了Cluster的整體架構(gòu)，接下來我們看看Cluster的啟動流程。以Application mode部署到Y(jié)arn為例(其他模式的啟動類似，只是啟動的主類不同)。該方式下的主類為：YarnApplicationClusterEntryPoint，其內(nèi)部調(diào)用了ClusterEntrypoint的方法，最終是通過ClusterEntrypoint類的runCluster()方法來創(chuàng)建DispatcherResourceManagerComponent對象。

DispatcherResourceManagerComponent

接下來我們看看DispatcherResourceManagerComponent中的具體屬性信息

    @Nonnull private final DispatcherRunner dispatcherRunner;
    @Nonnull private final ResourceManagerService resourceManagerService;
    @Nonnull private final RestService webMonitorEndpoint;
    @Nonnull private final LeaderRetrievalService dispatcherLeaderRetrievalService;
    @Nonnull private final LeaderRetrievalService resourceManagerRetrievalService;

Runner代碼

這里我們并沒有看到Dispatcher，而是一個(gè)類似名字的DispatcherRunner.DispatcherRunner是來管理Dispatcher如何運(yùn)行的。類似ResourceManagerService是來管理ResourceManager的生命周期的。

HA代碼框架

另外由于這些服務(wù)都有雙機(jī)容錯(cuò)機(jī)制(HA), 所以這里在看相關(guān)代碼的時(shí)候會產(chǎn)生一定的干擾，本篇的最后我們來介紹下這塊HA的相關(guān)的機(jī)制，這樣對大家來梳理相關(guān)的流程會更清晰。 Leader的選舉，是通過LeaderElectionService(選舉服務(wù)，實(shí)現(xiàn)類為DefaultLeaderElectionService)和LeaderContender(競選者)共同來完成的。具體過程為LeaderElectionService.start(LeaderContender)，啟動選舉服務(wù)，傳入LeaderContender信息，等選舉成功后，會回調(diào)LeaderContender的grantLeadership()方法，F(xiàn)link中相關(guān)的服務(wù)都實(shí)現(xiàn)了LeaderContender接口。所以理清這個(gè)邏輯后，我們在看到相關(guān)服務(wù)的start()方法中只調(diào)用了leaderElectionService.start方法時(shí)也不用懵了，直接看該服務(wù)的grantLeadership方法來梳理邏輯。 LeaderElectionDriver：進(jìn)行Leader的選舉和保存Leader的信息，具體的實(shí)現(xiàn)有ZooKeeperLeaderElectionDriver和KubernetesLeaderElectionDriver

那如何獲取Leader的地址呢，也提供了相應(yīng)的接口LeaderRetrievalService和LeaderRetrievalLister，啟動一個(gè)對Leader地址的監(jiān)聽，leader有變化時(shí)會得到通知。

總結(jié)

本篇我們了解了Flink的部署模式，按Job提交方式和一個(gè)集群可同時(shí)運(yùn)行任務(wù)的數(shù)量的不同，分為ApplicationMode和SessionMode2種模式。接著介紹了Cluster的整體架構(gòu)和啟動流程，主要包括Dispatcher、ResourceManager和WebMonitorEndpoint。最后介紹了HA處理的整體框架，便于大家更好的梳理核心流程。

以上就是Flink部署集群整體架構(gòu)源碼分析的詳細(xì)內(nèi)容，更多關(guān)于Flink部署集群架構(gòu)的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論