1、概述

• 把Flink应用提交给YARN的ResourceManager，从NodeManager上申请容器，创建JobManager和TaskManager
• 根据运行在JobManger上的Job的需要的插槽数量，动态地给TaskManager分配资源

Flink on YARN 架构图

2、下载和解压

https://archive.apache.org/dist/flink/

tar -zxvf flink-1.13.1-bin-scala_2.12.tgz -C $B_HOME/
cd $B_HOME
mv flink-1.13.1 flink
chown -R root:root flink

3、环境变量

https://yellow520.blog.csdn.net/article/details/112692486

# Flink
export FLINK_HOME=$B_HOME/flink
export HADOOP_CLASSPATH=`hadoop classpath`

4、三种部署模式

4.1、Session-Cluster（会话模式）

在YARN中初始化一个Flink集群，开辟指定的资源，以后提交任务都向这里提交
这个Flink集群会常驻在YARN集群中

4.1.1、开启会话

$FLINK_HOME/bin/yarn-session.sh \
-s 2 \
-jm 1024 \
-tm 1024 \
-nm a1 \
-d

参数	说明
-t
-s(--slots)	每个TaskManager的slot数量
-jm	JobManager的内存（单位MB）
-tm	每个TaskManager的内存（单位MB）
-nm	YARN上应用程序名字
-d	后台执行

8088端口

集群jps

4.1.2、在会话上运行jar

$FLINK_HOME/bin/flink run -c WordCount FlinkPractise-1.0-SNAPSHOT.jar

4.1.3、关闭会话

yarn application --list
yarn application --kill application_1636462413722_0003

4.2、Per-Job Cluster Mode（任务独立提交模式）

每次提交都会创建一个新的Flink集群，任务之间互相独立
任务执行完成之后创建的集群也会消失

$FLINK_HOME/bin/flink run \
-t yarn-per-job \
-nm a2 \
-c WordCount FlinkPractise-1.0-SNAPSHOT.jar

报错：
Exception in thread “Thread-5” java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration ‘classloader.check-leaked-classloader’.
解决：
yarn-site.xml中加入classloader.check-leaked-classloader: false

4.3、Application Mode

应用jar包的main函数在JobManager上执行；只要应用程序执行结束，Flink集群会马上被关闭，也可手动停止。

$FLINK_HOME/bin/flink run-application \
-t yarn-application \
-nm a3 \
-c WordCount FlinkPractise-1.0-SNAPSHOT.jar

区别	Per-Job-Cluster	Application Mode
main函数	在客户端执行（可以看到打印结果）	在JobManager中执行

生产建议使用Per-Job Cluster Mode或Application Mode

5、高可用

JobManager挂掉后，YARN会再次启动一个

vim $HADOOP_HOME/etc/hadoop/yarn-site.xml

<property>
 <name>yarn.resourcemanager.am.max-attempts</name>
 <value>4</value>
 <description>The maximum number of application master execution attempts.</description>
</property>

vim $FLINK_HOME/conf/flink-conf.yaml

yarn.application-attempts: 3
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop102:8020/flink/yarn/ha
high-availability.zookeeper.quorum: hadoop102:2181,hadoop103:2181,hadoop104:2181
high-availability.zookeeper.path.root: /flink-yarn