flink中怎么部署集群

MiniCluster

這種模式我們一般是在用IDE調(diào)試程序的時(shí)候用到,當(dāng)我們?cè)诒镜赜肐DE開發(fā)程序的時(shí)候，執(zhí)行main方法，flink會(huì)在本地啟動(dòng)一個(gè)包含jobmanager和taskmanager的進(jìn)程的minicluster，程序運(yùn)行完成之后，這個(gè)cluster進(jìn)程退出。

Standalone

這種模式就是直接在物理機(jī)上啟動(dòng)flink集群。我們可以通過{FLINK_HOME}/conf/flink-conf.yaml.
此外，我們可以用 ${FLINK_HOME}/bin/taskmanager.sh start 再啟動(dòng)一個(gè)taskmanager。

這時(shí)我們通過jps命令查看一下啟動(dòng)的進(jìn)程

76085 StandaloneSessionClusterEntrypoint
76331 TaskManagerRunner
76846 TaskManagerRunner

我們看到這時(shí)候啟動(dòng)了兩個(gè)taskmanager

 

這種部署模式對(duì)flink集群的資源管理是flink自己維護(hù)的，在生產(chǎn)環(huán)境下用的不多，所以我們也不做過多描述.

yarn

yarn session

• 部署在yarn集群上的flink集群都是把資源的管理交給了yarn來管理。
• yarn session的部署模式就是先預(yù)先在yarn集群上啟動(dòng)一個(gè)flink集群，我們可以把我們寫好的flink任務(wù)直接提到這個(gè)集群上。

啟動(dòng)集群的命令如下：

${FLINK_HOME}/bin/yarn-session.sh

這個(gè)命令有很多的參數(shù)，可以在后面加 -h 看下，我這里著重介紹一下 -d參數(shù)。
加上-d之后，指的是隔離模式，也就是啟動(dòng)之后和客戶端就斷了聯(lián)系，如果要停止集群，需要通過yarn application -kill {applicationId} 來停止集群.

• 如果沒指定-d，這種情況集群會(huì)和客戶端一直保持著連接，客戶端退出之后，集群也會(huì)退出。

• 提交任務(wù)
  往yarn session集群提交任務(wù)，只需要在相應(yīng)的客戶端機(jī)器上，通過${FLINK_HOME}/bin/flink run -d user.jar 這樣的命令就可以提交到session集群.

  此外我們還可以通過web ui最后一項(xiàng)來提交任務(wù)

• 這種session模式一般適用于批任務(wù)，也就是執(zhí)行一段時(shí)間以后可以終止的任務(wù)，因?yàn)閷?duì)于這種短時(shí)間執(zhí)行的任務(wù)，可以避免在申請(qǐng)資源方面浪費(fèi)過多時(shí)間。

• 集群?jiǎn)?dòng)之后，是沒有給flink集群分配資源的，當(dāng)提交任務(wù)之后，yarn集群會(huì)根據(jù)請(qǐng)求再給任務(wù)分配資源，任務(wù)執(zhí)行完成之后，系統(tǒng)隔一段時(shí)間會(huì)釋放相應(yīng)的資源.(這個(gè)時(shí)間是可配置的，為了防止馬上有任務(wù)又來了，重新申請(qǐng)資源)

yarn per job

• 我們上面講了session模式部署集群,這種模式可以在一個(gè)集群里跑很多的任務(wù)，這些任務(wù)共享了flink集群的資源，隔離性做的不是很好，所以flink還提供了另外一種執(zhí)行模式：yarn per job模式。

• 這種模式會(huì)在yarn上為每個(gè)flink任務(wù)都建立一個(gè)單獨(dú)的集群，優(yōu)勢(shì)就是每個(gè)任務(wù)單獨(dú)的進(jìn)行資源管理，和其他任務(wù)資源隔離。這種模式適用于對(duì)啟動(dòng)時(shí)間不太敏感，需要長(zhǎng)時(shí)間運(yùn)行的流任務(wù)。

• 啟動(dòng)命令

${FLINK_HOME}/bin/flink run -d -p 4 -ys 2 -m yarn-cluster -c com.example.Test userjar.jar arg1 arg2

提交成功之后，我們會(huì)在yarn的管理頁面看到一個(gè)類似的任務(wù)

 

這個(gè)啟動(dòng)命令也有很多的參數(shù)，我就不一一講解了，我用大白話講講我認(rèn)為最核心的幾個(gè)參數(shù)。

1. -d  采用分離模式
2. -p  程序的并行度
3. -ys 每個(gè)taskmanager有幾個(gè)slot,我們可以簡(jiǎn)單的理解為flink會(huì)把taskmanager的內(nèi)存分成幾份，在某些條件下，程序可以共用slot，提高效率，至于slot的概念，我們后續(xù)再講，今天就不多說了。用并行度除以這個(gè)值，然后就會(huì)得到flink會(huì)啟動(dòng)幾個(gè)taskmanager     ，所以為了避免有多余的slot，我們最好設(shè)置并行度除以這個(gè)ys值能整除。
4. -c 程序的入口類，我們可以在程序打包的時(shí)候指定入口類，如果沒有指定或者程序中有很多類，我們就需要通過這個(gè)-c參數(shù)來指定入口類了。
5. 在命令行最后的參數(shù)是用戶jar包的參數(shù).

• 停止命令    
  第一，我們可以在flink的頁面上通過停止flink的任務(wù)來停止集群，在我們停止了flink任務(wù)之后，yarn會(huì)自動(dòng)釋放相應(yīng)的資源。

 

第二，通過命令行來停止：

${FLINK_HOME}/bin/flink stop -m yarn-cluster -yid application_1592386606716_0005 c8ee546129e8480809ee62a4ce7dd91d

這個(gè)時(shí)候需要指定yarn applicationId和flink job id
第三，通過程序來停止

https://blog.csdn.net/zhangjun5965/article/details/106820591

如果我們做了一個(gè)實(shí)時(shí)平臺(tái)這樣的系統(tǒng)，就不能手工通過命令行來停止了，可以調(diào)用相應(yīng)的api來停止任務(wù).

• 啟動(dòng)流程    
  當(dāng)我們執(zhí)行完相應(yīng)的命令之后，系統(tǒng)會(huì)把flink的jar、相關(guān)的配置文件、用戶的jar都上傳到hdfs 的一個(gè)臨時(shí)目錄，默認(rèn)是/user/{USER}/.flink/{applicationId}, 然后再構(gòu)建flink集群的時(shí)候，再去找個(gè)目錄去獲取，程序部署成功之后，刪除相應(yīng)的臨時(shí)目錄

application模式

這種模式是在flink 1.11 版本中提供的，flink的yarn per job模式啟動(dòng)的時(shí)候會(huì)把本地的flink的jar和用戶的jar都上傳到hdfs，這個(gè)過程非常的消耗網(wǎng)絡(luò)的帶寬，如果同時(shí)有多個(gè)人提交任務(wù)的話，那么對(duì)網(wǎng)絡(luò)的影響就更大，此外，每次提交任務(wù)的時(shí)候flink的jar包是一樣的，也不用每次都拷來拷去的，所以flink提供了一種新的application模式，可以把flink的jar和用戶的jar都預(yù)先放到hdfs上，這樣就能省去yarn per job模式提交任務(wù)的jar包拷貝工作，節(jié)省了帶寬，加快了提交任務(wù)的速度.

具體的命令如下：

./bin/flink run-application -p 1 -d -t yarn-application \
-yD yarn.provided.lib.dirs="hdfs://localhost/data/flink/libs/" \
hdfs://localhost/data/flink/user-lib/TopSpeedWindowing.jar