Flink 作业提交

问题思考

1.Flink作业是如何提交到集群的？

2.Flink集群是如何在资源管理集群上启动起来的？

3.Flink的计算资源是如何分配给作业的？

4.Flink作业提交之后是如何启动的？

提交流程

作业可以选择Session和Per-Job模式两种集群

1. Session模式的集群，一个集群中运行多个作业
2. Per-Job模式的集群，一个集群只运行一个作业，作业执行完毕则集群销毁

 

根据Flink Client提交作业后是否可以退出Client进程，提交模式又可以分为Detached模式和Attached模式。Detached模式下，Flink lient创建完集群之后，可以退出命令行，集群独立运行。Attached模式下，Flink Client创建完集群后，不能关闭命令行窗口。

流水按执行器PipelineExecutor

PipelineExecutor是将Client生成JobGraph之后，将作业提交给集群的重要环节。Session和Per-Job这两种模式下，集群的启动时机、提交作业的方式不同，所以在及生产环境中有两种PipelineExecutor。Session模式对应于AbstractSessionClusterExecutor,Per-Jon模式对应于AbstractJobClusterExecutor

除了上述两种部署模式外，在IDE环境中运行Flink MiniCluster进行调试，使用LocalExecutor。

1.session模式

该模式下，作业共享集群资源，作业通过Http协议进行提交

standalone模式本质上就是一种session模式，所以不支持Per-Job模式。

   在Session模式下，Yarn作业提交使用yarn-session.sh脚本，k8s作业提交使用kubernetes-session.sh脚本。两者的具体实现不同，但逻辑是类似的，在启动脚本的时候就会检查是否存在已经启动好的Flink Session模式集群，如果没有，则启动一个Flink Session模式集群，然后在PipelineExecutor中，通过Dispatcher提供的Rest接口提交JobGraph,Dispatcher为每个作业启动一个JobMaster,进入作业执行阶段。

2.Per-Job模式

   该模式下，一个作业一个集群，作业之间相互隔离。

Per-Job模式下，因为不需要共享集群，所以在PipelineExecutor中执行作业提交的时候，可以创建集群并将JobGraph以及需要的文件等一同提交给Yarn集群，Yarn集群在容器中启动Flink Master进程(即JobManager进程)，进行一系列的初始化动作，初始化完毕之后，从文件系统中获取JobGraph，交给Dispatcher。之后的执行流程于Session模式下的执行流程相同。

Yarn-Session提交流程

从总体上来说，在Yarn集群上使用Session模式提交Flink作业的过程分为3个阶段

1. 在Yarn上启动Flink Session模式的集群
2. 其次通过Flink Client提交作业
3. 进行作业调度

1.启动集群

（1）使用bin/yarn-session.sh提交会话模式的作业

如果提交已经存在的集群，则获取Yarn集群信息、应用ID、并准备提交作业。

如果是启动新的Yarn Session集群，则进入步骤(2)

（2）Yarn启动新Flink集群

• 如果没有集群，则创建一个新的Session模式的集群。首先将应用配置(flink-conf.yaml、logback.xml、log4j.properties)和相关文件（flink jar、配置类文件、用户jar文件、JobGraph对象等）上传至分布式存储(如HDFS)的应用暂存目录。
• 通过Yarn Client向Yarn提交Flink创建集群的申请，Yarn分配资源，在申请的Yarn Container中初始化并启动Flink JobManager进程，在JobManager进程中运行YarnSessionClusterEntrypoint作为集群启动的入口(不同的集群部署模式有不同的ClusterEntrypoint实现),初始化Dispatcher、ResourceManager,启动相关的RPC服务，等待Client通过Rest接口提交作业。

2.作业提交

Yarn集群准备好后，开始作业提交

（1）Flink Client通过Rest向Dispatcher提交JobGraph

  (2)