Flink 支持 Standalone 独立部署和 YARN、Kubernetes、Mesos 等集群部署模式，其中 YARN 集群部署模式在国内的应用越来越广泛。Flink 社区将推出 Flink on YARN 应用解读系列文章，分为上、下两篇。本文基于 FLIP-6 重构后的资源调度模型将介绍 Flink on YARN 应用启动全流程，并进行详细步骤解析。
Flink on YARN 应用启动流程图
Flink on YARN 集群部署模式涉及 YARN 和 Flink 两大开源框架，应用启动流程的很多环节交织在一起，为了便于大家理解，在一张图上画出了 Flink on YARN 基础架构和应用启动全流程，并对关键角色和流程进行了介绍说明。整个启动流程被划分成客户端提交（流程标注为紫色）、Flink Cluster 启动和 Job 提交运行（流程标注为橙色）两个阶段分别阐述，由于分支和细节太多，本文会忽略掉一些，只介绍关键流程（基于 Flink 开源 1.9 版本源码整理）。
客户端提交流程
1、 执行命令:bin/flinkrun-d-myarn-cluster...或bin/yarn-session.sh...来提交per-job运行模式或session运行模式的应用；
2、 解析命令参数项并初始化，启动指定运行模式，如果是per-job运行模式将根据命令行参数指定的Job主类创建jobgraph；
- 如果可以从命令行参数(-yid
<APPLICATION\_ID>)或 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)和相关文件(flinkjars、shipfiles、userjars、jobgraph等)上传至分布式存储(例如HDFS)的应用暂存目录(/user/${user.name}/.flink/)；
5、 准备应用提交上下文(ApplicationSubmissionContext，包括应用的名称、类型、队列、标签等信息和应用Master的container的环境变量、classpath、资源大小等)，注册处理部署失败的shutdownhook（清理应用对应的HDFS目录），然后通过YarnClient向YARNRM提交应用；
6、 循环等待直到应用状态为RUNNING，包含两个阶段：；
- 循环等待应用提交成功（SUBMITTED）：默认每隔 200ms 通过 YarnClient 获取应用报告，如果应用状态不是 NEW 和 NEW_SAVING 则认为提交成功并退出循环，每循环 10 次会将当前的应用状态输出至日志："Application submission is not finished, submitted application
<APPLICATION\_ID>is still in<APP\_STATE>"，提交成功后输出日志："Submitted application<APPLICATION\_ID>" - 循环等待应用正常运行（RUNNING）：每隔 250 ms 通过 YarnClient 获取应用报告，每轮循环也会将当前的应用状态输出至日志："Deploying cluster, current state
<APP\_STATE>"。应用状态成功变为 RUNNING 后将输出日志"YARN application has been deployed successfully."并退出循环，如果等到的是非预期状态如 FAILED/FINISHED/KILLED，就会在输出 YARN 返回的诊断信息（"The YARN application unexpectedly switched to state<APP\_STATE>during deployment. Diagnostics from YARN: ..."）之后抛出异常并退出。
Flink Cluster 启动流程
1、 YARNRM中的ClientRMService（为普通用户提供的RPC服务组件，处理来自客户端的各种RPC请求，比如查询YARN集群信息，提交、终止应用等）接收到应用提交请求，简单校验后将请求转交给RMAppManager（YARNRM内部管理应用生命周期的组件）；
2、 RMAppManager根据应用提交上下文内容创建初始状态为NEW的应用，将应用状态持久化到RM状态存储服务（例如ZooKeeper集群，RM状态存储服务用来保证RM重启、HA切换或发生故障后集群应用能够正常恢复，后续流程中的涉及状态存储时不再赘述），应用状态变为NEW_SAVING；
3、 应用状态存储完成后，应用状态变为SUBMITTED；RMAppManager开始向ResourceScheduler（YARNRM可拔插资源调度器，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申请AMContainer（Container是YARN中资源的抽象，包含了内存、CPU等多维度资源），应用实例状态变为ACCEPTED；
7、 ResourceScheduler会根据优先级（队列/应用/请求每个维度都有优先级配置）从根队列开始层层递进，先后选择当前优先级最高的子队列、应用直至具体某个请求，然后结合集群资源分布等情况作出分配决策，AMContainer分配成功后，应用实例状态变为ALLOCATED_SAVING，并触发应用实例状态存储流程，存储成功后应用实例状态变为ALLOCATED；
8、 RMAppManager维护的应用实例开始通知ApplicationMasterLauncher（AM生命周期管理服务，负责启动或清理AMcontainer）启动AMcontainer，ApplicationMasterLauncher与YARNNodeManager（下文简称YARNNM，与YARNRM保持通信，负责管理单个节点上的全部资源、Container生命周期、附属服务等，监控节点健康状况和Container资源使用）建立通信并请求启动AMcontainer；
9、 ContainerManager（YARNNM核心组件，管理所有Container的生命周期）接收到AMcontainer启动请求，YARNNM开始校验ContainerToken及资源文件，创建应用实例和Container实例并存储至本地，结果返回后应用实例状态变为LAUNCHED；
10、 ResourceLocalizationService（资源本地化服务，负责Container所需资源的本地化它能够按照描述从HDFS上下载Container所需的文件资源，并尽量将它们分摊到各个磁盘上以防止出现访问热点）初始化各种服务组件、创建工作目录、从HDFS下载运行所需的各种资源至Container工作目录（路径为:${yarn.nodemanager.local-dirs}/usercache/${user}/appcache/<APPLICATION_ID>/<CONTAINER_ID>）；
11、 ContainersLauncher（负责container的具体操作，包括启动、重启、恢复和清理等）将待运行Container所需的环境变量和运行命令写到Container工作目录下的launch_container.sh脚本中，然后运行该脚本启动Container；
12、 Container进程加载并运行ClusterEntrypoint(FlinkJobManager入口类，每种集群部署模式和应用运行模式都有相应的实现，例如在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负责外部资源管理，与YARNRM建立通信并保持心跳，申请或释放TaskManager资源，注销应用等；SlotManager则负责内部资源管理，维护全部Slot信息和状态）及相关服务，创建异步AMRMClient，开始注册AM，注册成功后每隔一段时间（心跳间隔配置项：${yarn.heartbeat.interval}，默认5s）向YARNRM发送心跳来发送资源更新请求和接受资源变更结果YARNRM内部该应用和应用运行实例的状态都变为RUNNING，并通知AMLivelinessMonitor服务监控AM是否存活状态，当心跳超过一定时间（默认10分钟）触发AMfailover流程；
14、 启动Dispatcher（负责接收用户提供的作业，并且负责为这个新提交的作业拉起一个新的JobManager）及相关服务（包括RESTendpoint等），在per-job运行模式下，Dispatcher将直接从Container工作目录加载JobGraph文件；在session运行模式下，Dispatcher将在接收客户端提交的Job（_通过BlockServer接收jobgraph文件）后再进行后续流程；
15、 根据JobGraph启动JobManager（负责作业调度、管理Job和Task的生命周期），构建ExecutionGraph（JobGraph的并行化版本，调度层最核心的数据结构）；
16、 JobManager开始执行ExecutionGraph，向ResourceManager申请资源；
17、 ResourceManager将资源请求加入等待请求队列，并通过心跳向YARNRM申请新的Container资源来启动TaskManager进程；后续流程如果有空闲Slot资源，SlotManager将其分配给等待请求队列中匹配的请求，不用再通过18.YarnResourceManager申请新的TaskManager；
18、 YARNApplicationMasterService接收到资源请求后，解析出新的资源请求并更新应用请求信息；
19、 YARNResourceScheduler成功为该应用分配资源后更新应用信息，ApplicationMasterService接收到FlinkJobManager的下一次心跳时返回新分配资源信息；
20、 FlinkResourceManager接收到新分配的Container资源后，准备好TaskManager启动上下文（ContainerLauncherContext，生成TaskManager配置并上传至分布式存储，配置其他依赖和环境变量等），然后向YARNNM申请启动TaskManager进程，YARNNM启动Container的流程与AMContainer启动流程基本类似，区别在于应用实例在NM上已存在并未RUNNING状态时则跳过应用实例初始化流程，这里不再赘述；
21、 TaskManager进程加载并运行YarnTaskExecutorRunner（FlinkTaskManager入口类），初始化流程完成后启动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；
版权声明：本文不是「本站」原创文章，版权归原作者所有 | 原文地址: