Flink任务提交流程和运行模式

任务提交流程

Flink 的提交流程随着部署模式、资源管理平台的不同，会有不同的变化。这里做进一步的抽象，形成一个大概高视角的任务执行流程图，如下：

Flink按照集群和资源管理的划分运行模式有：Standalone、Flink On Yarn、K8S等。

Standalone

Standalone为独立模式，独立运行，不依赖外部资源调度管理框架。如果资源不足或出现故障，没有自动扩展和重分配的机制，需要手动处理。一般适合开发测试和作业较少的场景。支持的部署模式有：会话部署模式、应用部署模式。不支持PerJob（单作业）部署模式。

会话模式

首先启动集群，然后Web访问JobManager的8081端口提交任务或命令提交，提交任务如下：

cd ${FLINK_HOME}
bin/start-cluster.sh # 启动集群，根据配置文件TM的slot划分成静态的
bin/flink run -m master:8081 -c pers.xxm.flink.MyFlink /tmp/mytask.jar
bin/flink cancel <app_id> # id可通过flink list或UI查看
bin/stop-cluster.sh # 停止集群

再次提交第二个Job时，JobManager和TaskManager还是原来的进程，在JobManager内部会重新启动JobMaster线程，类似Spark的Driver。新的任务继续占用TaskManager的插槽，如果插槽不足任务提交失败。

应用模式

该模式不用启动集群。提交任务如下：

cd ${FLINK_HOME}
mv /tmp/mytask.jar lib/ # 将jar包放入lib目录
bin/standalone-job.sh start --job-classname pers.xxm.flink.MyFlink # JobManager机器执行
bin/taskmanager.sh start # 在所有需要跑TaskManager的机器执行
bin/taskmanager.sh stop # 停止集群，同时作业停止，集群销毁
bin/standalone-job.sh stop # 停止集群

该模式在8081端口也可以看到集群和任务运行。此时如果通过UI取消作业运行集群也会销毁。

Flink On Yarn

Flink集群安装在Hadoop集群上或者下载Flink依赖的Jar包，建议安装在Hadoop集群上，参照环境配置如下：

# 配置环境变量，所有机器
vim /etc/profile.d/my_env.sh # 内容如下4行
HADOOP_HOME=/usr/local/hadoop
export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
export HADOOP_CLASSPATH=`hadoop classpath`
source /etc/profile.d/my_env.sh
# 启动YARN集群，master运行，当前节点启动RM，在slaves配置的节点启动NM
start-yarn.sh
start-dfs.sh # 如果需要也可启动HDFS

YARN模式根据并行度除以每个TM插槽数向上取整动态申请TM，每个TM的slot参考Flink配置文件。该运行模式下支持的部署包括会话、单作业、应用模式。

会话模式

首先需要申请YARN会话，Yarn Session，然后启动Flink集群。启动会话应用如下：

cd $FLINK_HOME
# 执行后在YARN的8088端口UI查看生成了一个应用
bin/yarn-session.sh -nm my_app

此时在flink-conf.yaml的配置被覆盖即无效。启动后YARN会自动分配JM的机器和端口，在终端日志中会打印JM Web Interface地址，也可通过YARN的界面找到Tracking UI进入JM，这时使用YARN代理的方式进入。

未提交作业时，TM个数为0，因为Flink会根据运行在JM的作业所需slot动态分配TM。可提交多个作业。Job取消或结束后，占用的slot和tm会显示为可用，但过一会会被回收，显示总数和可用都为0。也可通过命令行提交任务到YARN会话，如下：

# 提交时查找/tmp/.yarn-properties-<username>文件，根据该文件对应YARN提交任务
bin/flink run -c pers.xxm.flink.MyFlink /tmp/mytask.jar
# 关闭YARN会话集群，该命令在启动上述会话时已经打印在标准输出中
echo "stop" | bin/yarn-session.sh -id <app_id>

单作业模式

每个作业占用一个YARN应用，即YARN集群，提交方式如下：

# 加上参数-d会推出占用模式在后台运行，CTRL+C退出不会影响集群的运行
bin/flink run -t yarn-per-job -c pers.xxm.flink.MyFlink /tmp/mytask.jar
bin/flink list -t yarn-per-job -Dyarn.application.id=<appid> # 查看集群中作业ID
bin/flink cancel -t yarn-per-job -Dyarn.application.id=<appid> <job_id> # 关闭上面的作业ID

在Flink的UI界面CANCEL任务后，YARN的应用状态变为FINISHED，这种关闭和上面命令行是一样的。

应用模式

Flink-1.11之后加入应用模式，和上个YARN单作业提交区别是此时提交的客户端不做代码解析等操作，这也是推荐的模式。提交方式如下：

bin/flink run-application -t yarn-application -c pers.xxm.flink.MyFlink /tmp/mytask.jar
bin/flink list -t yarn-application -Dyarn.application.id=<appid> # 查看集群中作业ID
bin/flink cancel -t yarn-application -Dyarn.application.id=<appid> <job_id> # 关闭作业ID

YARN模式优化

YARN的工作原理是每次执行任务时，都需要将Flink和用户的Jar包上传到HDFS上，所以在YARN的单作业和应用部署模式下，可以将依赖JAR包先上传到HDFS，然后指定路径，此时每次提交任务不会再次上传jar包到HDFS。

hadoop fs -mkdir /yarn/flink/dist # 创建目录
hadoop fs -mkdir /yarn/flink/jobs # 创建目录
hadoop fs -put lib/ /yarn/flink/dist # 将lib目录上传到dist目录下
hadoop fs -put plugins/ /yarn/flink/dist
hadoop fs -put /tmp/mytask.jar /yarn/flink/jobs # 自己的任务jar包放到jobs目录
# 此时以应用部署模式为例提交任务如下
bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirs="hdfs://master:8020/yarn/flink/dist" -c pers.xxm.flink.MyFlink hdfs://master:8020/yarn/flink/jobs/mytask.jar