本节讲解Flink on YARN集群模式部署，包括会话模式（Session）、单作业模式（Perjob）、应用模式（Application），最后讲解Flink on YARN模式下高可用设置。

Apache Hadoop YARN （Yet Another Resource Negotiator，另一种资源协调者）是一种新的通用资源调度系统，可为上层应用提供统一的资源管理和调度，它的引入为集群在利用率、资源统一管理和数据共享等方面带来了巨大好处。Spark、Flink、Hive、Tez等项目都可以使用YANR来进行资源调度管理。YARN是Hadoop的一部分，部署Flink YARN集群首先需要安装Hadoop，安装Hadoop的过程，请参考官网https://hadoop.apache.org/。

实际生成环境很少选择Flink独立集群模式，大多选择Flink on YARN、Flink on K8S。Flink借助于YARN可以动态地分配或者释放JobManager、TaskManager资源，根据运行的作业需要的Slot数量，Flink客户端与YANR交互，YARN通过自身的ResourceManager、NodeManger分配集群资源，YARN把Flink的JobManager、TaskManager部署到YARN容器里。这样做到了按需使用硬件资源，有作业运行就去申请，运行完作业后再返还计算资源给YARN资源管理器。

1.4.1 设置环境变量

部署Flink YARN集群需要添加Hadoop相关依赖jar包，早期版本的Flink提供了Pre-bundled Hadoop捆绑包，它打包了访问Hadoop需要的客户端类，然后把它放入Flink的lib目录。由于Pre-bundled Hadoop捆绑包容易引起 Jar包冲突，以及Hadoop版本迭代更替维护困难，现推荐方式：通过设置环境变量HADOOP_CLASSPATH，Flink启动会读取HADOOP_CLASSPATH，并把其HADOOP_CLASSPATH的值加入到Flink启动依赖项中。

编辑文件：vim /etc/profile，设置HADOOP_CLASSPATH，“hadoop classpath”是hadoop命令，classpath是参数，需要确保“hadoop classpath”能正确执行，提前是安装Hadoop并把$HADOOP_HOME/bin添加$PATH环境变量中。

export HADOOP_CLASSPATH=`hadoop classpath`
# 保存后执行source /etc/profile，刷新环境变量

1.4.2 理解Flink on YARN三种部署模式

Flink在YANR上支持三种部署模式，分别是应用模式（Application Mode）、单作业模式（Per-Job Mode）和会话模式（YARN Session）。生产环境推荐使用Application Mode或者Per-Job Mode，因为它们提供更好的应用隔离性。

1.4.2.1 会话模式

会话模式是提前初始化好的Flink集群并常驻在YARN服务中，用户提交的所有作业都在该Flink集群内运行。会话模式下只有一个JobManager，TaskManager根据作业需要的资源数量动态拉起。由于多个作业会共用TaskManager，如果有一个TaskManager宕机，上面运行的所有作业都会失败，另外作业越多JobManager任务越重，容易成为系统瓶颈，实际使用中容易僵死。会话模式适合部署延迟非常敏感但运行时长较短的作业。

1.4.2.2 单作业模式

单作业模式下，每个作业提交到YARN都会启动一个Flink集群，JobManager、TaskManager都是专属某个作业的，作业之间相互隔离，单个作业的失败不影响其他作业。由于每个作业都需要全新拉起一个Flink集群，导致作业启动延迟较长。单作业模式下Flink集群的生命周期和作业周期保持一致，即是作业启动，Flink集群随之启动，作业结束，Flink集群也跟着销毁。单作业模式适合部署长时间运行的作业。

1.4.2.3 应用模式

应用模式是Flink 1.11新增的一种部署模式。它也是每个作业都启动一个Flink集群，区别于单作业模式，应用模式运行main()方法的位置不一样。客户端提交作业需要经历解析（运行main方法）、提取算子并生成JobGraph，接着提交JobGraph和Flink依赖库。我们用于提交作业的服务器往往是固定的机器，如果作业提交很频繁会导致客户端负载过大，以及大量的数据传输导致网络带宽占用。Flink为了优化单作业模式，把作业解析、提取、生成JobGraph的阶段放到了JobManager，这样避免客户端负载过大。JobManager是YARN动态拉取，且运行在不同YANR NodeManager上，这样能很好地把压力分散到各计算节点上。另外依赖的资源包通过指定HDFS路径，JobManager去HDFS拉取，一般HDFS的数据会分散到三个不同节点，且NodeManager会缓存数据，这儿有效规避了客户端网络带宽占用问题。

1.4.3 启动Flink YARN集群

1.4.3.1 会话模式

1.启动YARN Session集群，--detached参数代表后台运行。

./bin/yarn-session.sh --detached

启动成功后会有类似如下提示信息，可以看到JobManager Web地址、YANR上应用id、以及怎么优雅地关闭Flink应用。JobManager Web地址也可以通过YARN ResourceManager页面进入，使用方法：首先在YARN Applications列表找到目标应用，然后点击进入详细页面，接着找到 “Tracking URL: ApplicationMaster” ，点击就能看到JobManager Web地址了。

2022-01-16 16:25:17,301 INFO org.apache.flink.yarn.YarnClusterDescriptor                 [] - YARN application has been deployed successfully.
2022-01-16 16:25:17,302 INFO org.apache.flink.yarn.YarnClusterDescriptor                 [] - Found Web Interface host1.test.com:39344 of application 'application_1640606693101_0012'.
JobManager Web Interface: http://host1.test.com:39344
2022-01-16 16:25:17,531 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli               [] - The Flink YARN session cluster has been started in detached mode. In order to stop Flink gracefully, use the following command:
$ echo "stop" | ./bin/yarn-session.sh -id application_1640606693101_0012
If this should not be possible, then you can also kill Flink via YARN's web interface or via:
$ yarn application -kill application_1640606693101_0012
Note that killing Flink might not clean up all job artifacts and temporary files.

2.提交Flink作业到YANR集群

# 通过flink run执行提交作业，--detached代表后台运行
./bin/flink run --detached ./examples/streaming/TopSpeedWindowing.jar

注意：在Flink on YARN集群模式下，jobmanager.rpc.address参数会失效。JobManager由YARN动态拉起并分配到某计算节点，所以不能再通过配置的jobmanager.rpc.address地址访问该服务。

1.4.3.2 单作业模式

通过Per-Job模式提交Flink作业，Flink集群和作业一起启动。相比YARN Session下提交作业多了target参数：“-t yarn-per-job”。

./bin/flink run -t yarn-per-job --detached ./examples/streaming/TopSpeedWindowing.jar

查看运行的作业以及取消执行

# 列举正在执行的作业
./bin/flink list -t yarn-per-job -Dyarn.application.id=application_XXXX_YY
# 取消正在执行的作业，Per-Job模式下取消作业会导致Flink集群一并销毁
./bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_XXXX_YY <jobId>

1.4.3.3 应用模式

通过应用模式提交Flink作业，用户jar的main()方法在JobManager中执行，同样取消作业会导致Flink集群一并销毁。

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

为了避免客户端网络传输到Flink集群，通过yarn.provided.lib.dirs参数指定Flink依赖jar包，用户提前把Flink依赖jar包、用户应用jar包上传到HDFS集群。

# 指定包获取路径，而非本地客户端上传
./bin/flink run-application -t yarn-application \
-Dyarn.provided.lib.dirs="hdfs://myhdfs/my-remote-flink-dist-dir" \
hdfs://myhdfs/jars/my-application.jar

1.4.4 Flink on YARN高可用

在Flink on Yarn集群模式下，Flink高可用和独立模式有些差异，YANR资源管理框架可以动态拉起Flink集群，包含其中的JobManager。Flink on YARN集群的高可用只需要运行一个JobManager实例，当这个JobManager崩溃后，Yarn会重新启动一个新的JobManager。但完成这些动作的前提是作业信息需要保存到zookeeper。

Flink on YARN高可用依赖zookeeper，在配置Flink高可用前需先安装zookeeper集群。配置zookeeper参数的方式和独立集群配置一样，除此之外有个yarn.application-attempts参数需要关注。

1.编辑flink-conf.yaml配置文件，设置基于zookeeper的高可用。Flink on YARN修改了配置文件，不需要同步到其他节点，执行命令拉起Flink集群的时候客户端会把配置文件一并发送到ApplicationMaster。

#必填参数
high-availability: zookeeper  
#必填参数，多quorum格式：host1:port1,host2:port2,host3:port3
high-availability.zookeeper.quorum: host1.test.com:2181
#必填参数，storageDir需要是所有节点都能访问的地址，比如HDFS, S3, Ceph, nfs。如果使用hdfs地址则需要引入hadoop相关包，另外flink的启动用户必须有权限创建hdfs目录
high-availability.storageDir: hdfs:///flink/recovery

#可选参数，存储状态的zookeeper路径
high-availability.zookeeper.path.root: /flink
#应用尝试启动的次数，不能超过YARN设置的ApplicationMaster最大尝试次数
yarn.application-attempts: 2

应用尝试启动的次数yarn.application-attempts，2表示总共能重启的次数，包含作业最开始启动的那一次，当2次启动都失败时，YARN会认为这个作业失败了。需要注意的是，此处配置的上限是YANR服务中的yarn.resourcemanager.am.max-attempts参数，如果需要调高此处参数，则需要检查yarn.resourcemanager.am.max-attempts参数值。

2.ApplicationMaster最大执行次数，配置项在yarn-site.xml文件中。

<!-- YARN ApplicationMaster最大的执行次数. 默认值是2 -->
<property>
 <name>yarn.resourcemanager.am.max-attempts</name>
 <value>2</value>
</property>

3.不同版本的YARN在作业发生错误后，在关闭Container时采取的方式不尽相同。

- 在YARN 2.3.0 到 2.4.0版本中：当Application Master失败时，所有的Containers都将重启。

- 在YARN 2.4.0 到 2.6.0版本中：当Application Master失败时，TaskManager Containers仍会保持存活状态，这样做的好处是启动时间更快，并且不必等待再次获得容器资源。

- 在YARN 高于或等于2.6.0版本中：在以前的基础上设置了一个间隔时间，这个间隔时间的值等于Flink的Akka超时时间，当作业发生错误后，会以这个时间间隔来重启任务，当重试次数超过Flink设置的yarn.application-attempts值时，作业才会被系统kill。这样可以避免一个长时间失败的任务耗尽Application Master的尝试重启次数。

如果有任何疑问欢迎留言，笔者头条号与公众号同名：小圈数据