1。4FlinkonYARN集群部署Flink核心源码解读

　　本节讲解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 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.resourcemanager.am.max-attempts  2 
　　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的尝试重启次数。
　　如果有任何疑问欢迎留言，笔者头条号与公众号同名：小圈数据