Flink On Yarn模式

原理

為什么使用Flink On Yarn?

在實際開發(fā)中，使用Flink時，更多的使用方式是Flink On Yarn模式，原因如下：

-1.Yarn的資源可以按需使用，提高集群的資源利用率
-2.Yarn的任務(wù)有優(yōu)先級，根據(jù)優(yōu)先級運行作業(yè)
-3.基于Yarn調(diào)度系統(tǒng)，能夠自動化地處理各個角色的 Failover(容錯)
○ JobManager 進程和 TaskManager 進程都由 Yarn NodeManager 監(jiān)控
○ 如果 JobManager 進程異常退出，則 Yarn ResourceManager 會重新調(diào)度 JobManager 到其他機器
○ 如果 TaskManager 進程異常退出，JobManager 會收到消息并重新向 Yarn ResourceManager 申請資源，重新啟動 TaskManager

Flink如何和Yarn進行交互?

1.Client上傳jar包和配置文件到HDFS集群上
2.Client向Yarn ResourceManager提交任務(wù)并申請資源
3.ResourceManager分配Container資源并啟動ApplicationMaster,然后AppMaster加載Flink的Jar包和配置構(gòu)建環(huán)境,啟動JobManager

JobManager和ApplicationMaster運行在同一個container上。
一旦他們被成功啟動，AppMaster就知道JobManager的地址(AM它自己所在的機器)。
它就會為TaskManager生成一個新的Flink配置文件(他們就可以連接到JobManager)。
這個配置文件也被上傳到HDFS上。
此外，AppMaster容器也提供了Flink的web服務(wù)接口。
YARN所分配的所有端口都是臨時端口，這允許用戶并行執(zhí)行多個Flink

4.ApplicationMaster向ResourceManager申請工作資源,NodeManager加載Flink的Jar包和配置構(gòu)建環(huán)境并啟動TaskManager
5.TaskManager啟動后向JobManager發(fā)送心跳包，并等待JobManager向其分配任務(wù)

兩種方式

Session模式

特點：需要事先申請資源，啟動JobManager和TaskManger
優(yōu)點：不需要每次遞交作業(yè)申請資源，而是使用已經(jīng)申請好的資源，從而提高執(zhí)行效率
缺點：作業(yè)執(zhí)行完成以后，資源不會被釋放，因此一直會占用系統(tǒng)資源
應(yīng)用場景：適合作業(yè)遞交比較頻繁的場景，小作業(yè)比較多的場景

Per-Job模式

特點：每次遞交作業(yè)都需要申請一次資源
優(yōu)點：作業(yè)運行完成，資源會立刻被釋放，不會一直占用系統(tǒng)資源
缺點：每次遞交作業(yè)都需要申請資源，會影響執(zhí)行效率，因為申請資源需要消耗時間
應(yīng)用場景：適合作業(yè)比較少的場景、大作業(yè)的場景

操作

vim /export/server/hadoop/etc/hadoop/yarn-site.xml
添加：

<!-- 關(guān)閉yarn內(nèi)存檢查 --><property><name>yarn.nodemanager.pmem-check-enabled</name><value>false</value></property><property><name>yarn.nodemanager.vmem-check-enabled</name><value>false</value></property>

說明:
是否啟動一個線程檢查每個任務(wù)正使用的虛擬內(nèi)存量，如果任務(wù)超出分配值，則直接將其殺掉，默認(rèn)是true。
在這里面我們需要關(guān)閉，因為對于flink使用yarn模式下，很容易內(nèi)存超標(biāo)，這個時候yarn會自動殺掉job

2.同步

scp -r /export/server/hadoop/etc/hadoop/yarn-site.xml node2:/export/server/hadoop/etc/hadoop/yarn-site.xml scp -r /export/server/hadoop/etc/hadoop/yarn-site.xml node3:/export/server/hadoop/etc/hadoop/yarn-site.xml

3.重啟yarn

/export/server/hadoop/sbin/stop-yarn.sh /export/server/hadoop/sbin/start-yarn.sh

測試

Session模式

yarn-session.sh(開辟資源) + flink run(提交任務(wù))

1.在yarn上啟動一個Flink會話，node1上執(zhí)行以下命令

/export/server/flink/bin/yarn-session.sh -n 2 -tm 800 -s 1 -d

說明:

申請2個CPU、1600M內(nèi)存# -n 表示申請2個容器，這里指的就是多少個taskmanager # -tm 表示每個TaskManager的內(nèi)存大小 # -s 表示每個TaskManager的slots數(shù)量 # -d 表示以后臺程序方式運行

注意:

該警告不用管 WARN org.apache.hadoop.hdfs.DFSClient - Caught exception java.lang.InterruptedException

2. 查看UI界面

http://node1:8088/cluster

3.使用flink run提交任務(wù)：

/flink/bin/flink run /export/server/flink/examples/batch/WordCount.jar run /export/server/flink/examples/batch/WordCount.jar

4.通過上方的ApplicationMaster可以進入Flink的管理界面

5.關(guān)閉yarn-session：

yarn application -kill application_1599402747874_0001

rm -rf /tmp/.yarn-properties-root

Per-Job分離模式

1.直接提交job

/export/server/flink/bin/flink run -m yarn-cluster -yjm 1024 -ytm 1024 /export/server/flink/examples/batch/WordCount.jar # -m jobmanager的地址 # -yjm 1024 指定jobmanager的內(nèi)存信息 # -ytm 1024 指定taskmanager的內(nèi)存信息

2.查看UI界面

http://node1:8088/cluster

3.注意：

在之前版本中如果使用的是flink on yarn方式，想切換回standalone模式的話，如果報錯需要刪除：【/tmp/.yarn-properties-root】

rm -rf /tmp/.yarn-properties-root

因為默認(rèn)查找當(dāng)前yarn集群中已有的yarn-session信息中的jobmanager

總結(jié)

以上是生活随笔為你收集整理的Flink On Yarn模式，为什么使用Flink On Yarn?Session模式、Per-Job模式、关闭yarn的内存检查，由Yarn模式切换回standalone模式时需要注意的点的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。