4月20日，袋鼠云数栈技术研发团队工程师兰洋（花名：莫问）为大家直播分享《Flink提交流程&如何debug和跟踪流程（on yarn）》。错过直播的朋友可以钉钉扫描文末的二维码，加入钉钉群回看直播，或者在b站搜索“袋鼠云”观看视频。

b站视频网址：

https://www.bilibili.com/video/BV13K4y1P7ri

下面带大家来回顾下本次直播的内容，本次直播莫问大佬主要从以下几个方面来为大家进行分享。

在我们平常flink的开发中，通常是在ide中开发并使用flink的单机模式进行测试，测试完毕后会将代码打成jar文件然后通过Flink CLI或者Web UI提交作业到flink集群。但是在真正执行我们flink的作业之前，我们是否有考虑过以下几个问题：

• Flink作业是如何提交到集群的？
• Flink集群是如何在资源管理集群(yarn、k8s)上启动起来的？
• Flink到计算资源是如何分配给作业的？
• Flink作业提交之后是如何运行的？

1、Flink客户端提交流程
我们的Flink作业在开发完毕后，是需要提交到Flink集群执行。那么CliFrontend的main()方法便是提交作业到Flink集群的入口。其main()方法中主要的逻辑是触发我们打包好的jar中的main方法，然后交给PipelineExecutor#execute方法，最终会实例化一个具体的PipelineExecutor进行执行。但是在Application模式，跟Perjob以及Session模式有所区别，是通过ApplicationDeployer来进行任务的部署。

其过程如下图所示：



2、FlinkStreamSql客户端提交流程

FlinkX以及FlinkStreamSql的提交代码在项目的Launch模块下，下图是它们客户端的提交流程：

其实现关系如下：

1）使用./bin/start-cluster.sh启动flink集群。

2）Yarn启动Flink集群，其集群的启动入口是YarnSessionClusterEntryPoint。其启动集群的的步骤可以分为两部分。

第一部分是客户端会将应用配置(即flink-conf.yaml、logback.xml以及log4j.properties)和相关文件(flink jar、配置类文件、用户jar文件以及JobGraph对象等等)上传到分布式存储的应用暂存目录(.flink)。

第二部分是通过YarnClient向Yarn提交Flink创建集群的申请，Yarn分配容器资源，并且在申请的container中初始化并启动Flink JobManager进程，然后会在JobManager进程中运行YarnJobClusterEntryPoint作为集群的启动入口。初始化Dispatcher、ResourceManager以及RestEndPoint服务，并启动相关的RPC服务。

2、作业提交

集群准备好后，FlinkClient会向Dispathcher提交任务的JobGraph。Dispatcher获取到JobGraph后会为该JobGraph创建一个JobMaster,随后由JobMaster来负责作业调度、管理作业和Task的生命周期，并构建ExecutionGraph(JobGraph的并行化版本)。

1、参数配置

# jobmanager debug端口env.java.opts.jobmanager: "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5006"# taskmanager debug端口env.java.opts.taskmanager: "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005"

2、参数说明

• transport：指定调试数据的传送方式。dt_socket是指用SOCKET模式，另有dt_shmem指用共享内存方式，其中，dt_shmem只适用于windows平台。
  

• server：指是否支持在server模式的VM中。
  

• suspend: 指定是否在调试客户端建立起来后，再执行JVM。
  

• address: socket连接的端口号。

3、注意点

• 预先设置断点
  

• 保持远端代码跟所要调试的代码一致
  

• 线程并行调试。不要影响其他线程运行，可在断点出勾选Thread。