简介: 本文基于FLIP-6重构后的资源调度模型介绍Flink on YARN应用启动全流程,解答客户端和Flink Cluster的常见问题,分享相关问题的排查思路。
作者:杨弢(搏远)
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临时文件(
j
a
v
a
.
i
o
.
t
m
p
d
i
r
/
.
y
a
r
n
−
p
r
o
p
e
r
t
i
e
s
−
{java.io.tmpdir}/.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工作目录(路径为:
y
a
r
n
.
n
o
d
e
m
a
n
a
g
e
r
.
l
o
c
a
l
−
d
i
r
s
/
u
s
e
r
c
a
c
h
e
/
{yarn.nodemanager.local-dirs}/usercache/
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://github.com/apache/flink/tree/release-1.9.0?spm=a2c6h.12873639.0.0.209a6c7a2v1WeC&file=release-1.9.0
https://ci.apache.org/projects/flink/flink-docs-release-1.9/
https://github.com/apache/hadoop/tree/branch-3.2
http://hadoop.apache.org/docs/r3.2.0/