摘要：Flink 支持 Standalone 独立部署和 YARN、Kubernetes、Mesos 等集群部署模式，其中 YARN 集群部署模式在国内的应用越来越广泛。Flink 社区将推出 Flink

Flink 支持 Standalone 独立部署和 YARN、Kubernetes、Mesos 等集群部署模式，其中 YARN 集群部署模式在国内的应用越来越广泛。Flink 社区将推出 Flink

Flink 支持 Standalone 独立部署和 YARN、Kubernetes、Mesos 等集群部署模式，其中 YARN 集群部署模式在国内的应用越来越广泛。Flink 社区将推出 Flink on YARN 应用解读系列文章，分为上、下两篇。本文基于 FLIP-6 重构后的资源调度模型将介绍 Flink on YARN 应用启动全流程，并进行详细步骤解析。下篇将根据社区大群反馈，解答客户端和Flink Cluster的常见问题，分享相关问题的排查思路。

Flink on YARN 流程图

Flink on YARN集群部署模式涉及YARN和Flink两大开源框架，应用启动流程的很多环节交织在一起，为了便于大家理解，在一张图上画出了Flink on YARN基础架构和应用启动全流程，并对关键角色和流程进行了介绍说明，整个启动流程又被划分成客户端提交（流程标注为紫色）、Flink Cluster启动和Job提交运行（流程标注为橙色）两个阶段分别阐述，由于分支和细节太多，本文会忽略掉一些，只介绍关键流程（基于Flink开源1.9版本源码整理）。

客户端提交流程

1.执行命令:bin/flink run -d -m yarn-cluster ...或bin/yarn-session.sh ...来提交per-job运行模式或session运行模式的应用；

2.解析命令参数项并初始化，启动指定运行模式，如果是per-job运行模式将根据命令行参数指定的Job主类创建job graph；

• 如果可以从命令行参数(-yid )或YARN properties临时文件(${java.io.tmpdir}/.yarn-properties-${user.name})中获取应用ID，向指定的应用提交Job；
• 否则当命令行参数中包含 -d（表示detached模式）和 -m yarn-cluster（表示指定YARN集群模式），启动per-job运行模式；
• 否则当命令行参数项不包含 -yq（表示查询YARN集群可用资源）时，启动session运行模式；

3.获取YARN集群信息、新应用ID并启动运行前检查；

• 通过YarnClient向YARN ResourceManager(下文缩写为：YARN RM，YARN Master节点，负责整个集群资源的管理和调度)请求创建一个新应用（YARN RM收到创建应用请求后生成新应用ID和container申请的资源上限后返回），并且获取YARN Slave节点报告（YARN RM返回全部slave节点的ID、状态、rack、http地址、总资源、已使用资源等信息）；
• 运行前检查：(1) 简单验证YARN集群能否访问；(2) 最大node资源能否满足flink JobManager/TaskManager vcores资源申请需求；(3) 指定queue是否存在(不存在也只是打印WARN信息，后续向YARN提交时排除异常并退出)；(4)当预期应用申请的Container资源会超出YARN资源限制时抛出异常并退出；(5) 当预期应用申请不能被满足时（例如总资源超出YARN集群可用资源总量、Container申请资源超出NM可用资源最大值等）提供一些参考信息。

4.将应用配置(flink-conf.yaml、logback.xml、log4j.properties)和相关文件(flink jars、ship files、user jars、job graph等)上传至分布式存储(例如HDFS)的应用暂存目录(/user/${user.name}/.flink/)；

5.准备应用提交上下文(ApplicationSubmissionContext，包括应用的名称、类型、队列、标签等信息和应用Master的container的环境变量、classpath、资源大小等)，注册处理部署失败的shutdown hook（清理应用对应的HDFS目录），然后通过YarnClient向YARN RM提交应用；

6.循环等待直到应用状态为RUNNING，包含两个阶段：

• 循环等待应用提交成功（SUBMITTED）：默认每隔200ms通过YarnClient获取应用报告，如果应用状态不是NEW和NEW_SAVING则认为提交成功并退出循环，每循环10次会将当前的应用状态输出至日志："Application submission is not finished, submitted application is still in "，提交成功后输出日志："Submitted application "
• 循环等待应用正常运行（RUNNING）：每隔250ms通过YarnClient获取应用报告，每轮循环也会将当前的应用状态输出至日志："Deploying cluster, current state "。应用状态成功变为RUNNING后将输出日志"YARN application has been deployed successfully." 并退出循环，如果等到的是非预期状态如FAILED/FINISHED/KILLED,就会在输出YARN返回的诊断信息（"The YARN application unexpectedly switched to state during deployment. Diagnostics from YARN: ..."）之后抛出异常并退出。

Flink Cluster启动流程

1.YARN RM中的ClientRMService（为普通用户提供的RPC服务组件，处理来自客户端的各种RPC请求，比如查询YARN集群信息，提交、终止应用等）接收到应用提交请求，简单校验后将请求转交给RMAppManager（YARN RM内部管理应用生命周期的组件）；

2.RMAppManager根据应用提交上下文内容创建初始状态为NEW的应用，将应用状态持久化到RM状态存储服务（例如ZooKeeper集群，RM状态存储服务用来保证RM重启、HA切换或发生故障后集群应用能够正常恢复，后续流程中的涉及状态存储时不再赘述），应用状态变为NEW_SAVING；

3.应用状态存储完成后，应用状态变为SUBMITTED；RMAppManager开始向ResourceScheduler（YARN RM可拔插资源调度器，YARN自带三种调度器FifoScheduler/FairScheduler/CapacityScheduler，其中CapacityScheduler支持功能最多使用最广泛，FifoScheduler功能最简单基本不可用，今年社区已明确不再继续支持FairScheduler，建议已有用户迁至CapacityScheduler）提交应用，如果无法正常提交（例如队列不存在、不是叶子队列、队列已停用、超出队列最大应用数限制等）则抛出拒绝该应用，应用状态先变为FINAL_SAVING触发应用状态存储流程并在完成后变为FAILED；如果提交成功，应用状态变为ACCEPTED；

4.开始创建应用运行实例(ApplicationAttempt，由于一次运行实例中最重要的组件是ApplicationMaster，下文简称AM，它的状态代表了ApplicationAttempt的当前状态，所以ApplicationAttempt实际也代表了AM)，初始状态为NEW；

5.初始化应用运行实例信息，并向ApplicationMasterService（AM&RM协议接口服务，处理来自AM的请求，主要包括注册和心跳）注册，应用实例状态变为SUBMITTED；

6.RMAppManager维护的应用实例开始初始化AM资源申请信息并重新校验队列，然后向ResourceScheduler申请AM Container（Container是YARN中资源的抽象，包含了内存、CPU等多维度资源），应用实例状态变为ACCEPTED；

7.ResourceScheduler会根据优先级（队列/应用/请求每个维度都有优先级配置）从根队列开始层层递进，先后选择当前优先级最高的子队列、应用直至具体某个请求，然后结合集群资源分布等情况作出分配决策，AM Container分配成功后，应用实例状态变为ALLOCATED_SAVING，并触发应用实例状态存储流程，存储成功后应用实例状态变为ALLOCATED；

8.RMAppManager维护的应用实例开始通知ApplicationMasterLauncher（AM生命周期管理服务，负责启动或清理AM container）启动AM container，ApplicationMasterLauncher与YARN NodeManager（下文简称YARN NM，与YARN RM保持通信，负责管理单个节点上的全部资源、Container生命周期、附属服务等，监控节点健康状况和Container资源使用）建立通信并请求启动AM container；

9.ContainerManager（YARN NM核心组件，管理所有Container的生命周期）接收到AM container启动请求，YARN NM开始校验Container Token及资源文件，创建应用实例和Container实例并存储至本地，结果返回后应用实例状态变为LAUNCHED；

10.ResourceLocalizationService（资源本地化服务，负责Container所需资源的本地化。它能够按照描述从HDFS上下载Container所需的文件资源，并尽量将它们分摊到各个磁盘上以防止出现访问热点）初始化各种服务组件、创建工作目录、从HDFS下载运行所需的各种资源至Container工作目录（路径为: ${yarn.nodemanager.local-dirs}/usercache/${user}/appcache//）；

11.ContainersLauncher（负责container的具体操作，包括启动、重启、恢复和清理等）将待运行Container所需的环境变量和运行命令写到Container工作目录下的launch_container.sh脚本中，然后运行该脚本启动Container；

12.Container进程加载并运行ClusterEntrypoint(Flink JobManager入口类，每种集群部署模式和应用运行模式都有相应的实现，例如在YARN集群部署模式下，per-job应用运行模式实现类是YarnJobClusterEntrypoint，session应用运行模式实现类是YarnSessionClusterEntrypoint)，首先初始化相关运行环境：

• 输出各软件版本及运行环境信息、命令行参数项、classpath等信息；
• 注册处理各种SIGNAL的handler:记录到日志
• 注册JVM关闭保障的shutdown hook：避免JVM退出时被其他shutdown hook阻塞
• 打印YARN运行环境信息：用户名
• 从运行目录中加载flink conf
• 初始化文件系统
• 创建并启动各类内部服务（包括RpcService、HAService、BlobServer、HeartbeatServices、MetricRegistry、ExecutionGraphStore等）
• 将RPC address和port更新到flink conf配置

13.启动ResourceManager（Flink资源管理核心组件，包含YarnResourceManager和SlotManager两个子组件，YarnResourceManager负责外部资源管理，与YARN RM建立通信并保持心跳，申请或释放TaskManager资源，注销应用等；SlotManager则负责内部资源管理，维护全部Slot信息和状态）及相关服务，创建异步AMRMClient，开始注册AM，注册成功后每隔一段时间（心跳间隔配置项：${yarn.heartbeat.interval}，默认5s）向YARN RM发送心跳来发送资源更新请求和接受资源变更结果。YARN RM内部该应用和应用运行实例的状态都变为RUNNING，并通知AMLivelinessMonitor服务监控AM是否存活状态，当心跳超过一定时间（默认10分钟）触发AM failover流程；

14.启动Dispatcher（负责接收用户提供的作业，并且负责为这个新提交的作业拉起一个新的 JobManager）及相关服务（包括REST endpoint等），在per-job运行模式下，Dispatcher将直接从Container工作目录加载JobGraph文件；在session运行模式下，Dispatcher将在接收客户端提交的Job（_通过BlockServer接收job graph文件）后再进行后续流程；

15.根据JobGraph启动JobManager（负责作业调度、管理Job和Task的生命周期），构建ExecutionGraph（JobGraph的并行化版本，调度层最核心的数据结构）；

16.JobManager开始执行ExecutionGraph，向ResourceManager申请资源；

17.ResourceManager将资源请求加入等待请求队列，并通过心跳向YARN RM申请新的Container资源来启动TaskManager进程；后续流程如果有空闲Slot资源，SlotManager将其分配给等待请求队列中匹配的请求，不用再通过18. YarnResourceManager申请新的TaskManager；

**18.YARN ApplicationMasterService接收到资源请求后，解析出新的资源请求并更新应用请求信息；

19.YARN ResourceScheduler成功为该应用分配资源后更新应用信息，ApplicationMasterService接收到Flink JobManager的下一次心跳时返回新分配资源信息；

20.Flink ResourceManager接收到新分配的Container资源后，准备好TaskManager启动上下文（ContainerLauncherContext，生成TaskManager配置并上传至分布式存储，配置其他依赖和环境变量等），然后向YARN NM申请启动TaskManager进程，YARN NM启动Container的流程与AM Container启动流程基本类似，区别在于应用实例在NM上已存在并未RUNNING状态时则跳过应用实例初始化流程，这里不再赘述；

21.TaskManager进程加载并运行YarnTaskExecutorRunner（Flink TaskManager入口类），初始化流程完成后启动TaskExecutor（负责执行Task相关操作）；

22.TaskExecutor启动后先向ResourceManager注册，成功后再向SlotManager汇报自己的Slot资源与状态；

SlotManager接收到Slot空闲资源后主动触发Slot分配，从等待请求队列中选出合适的资源请求后，向

TaskManager请求该Slot资源

23.TaskManager收到请求后检查该Slot是否可分配（不存在则返回异常信息）、Job是否已注册（没有则先注册再分配Slot），检查通过后将Slot分配给JobManager；

24.JobManager检查Slot分配是否重复，通过后通知Execution执行部署task流程，向TaskExecutor提交task；

TaskExecutor启动新的线程运行Task。

原文链接： https://yq.aliyun.com/articles/719262?utm_content=g_1000078167

本文为云栖社区原创内容，未经允许不得转载。