Flink任务提交过程源码解析

在Flink内提交任务时，命令行命令入下：

flink run -c com.qihoo.anti.bootup coverage-stat-flink-1.0.jar

通过代码定位，发现CliFrontend是入口类；

首先，在指定的目录下，寻找配置文件，加载全局配置；同时，对commandline进行解析，在parseParameters函数内，根据action，执行不同的代码分支；在这里，主要关注ACTION_RUN的代码分支；

跟进看run方法，在此方法内，有两个函数，可以重点关注一下，一个是buildProgram函数，另一个是runProgram函数；

在buildProgram方法内，通过将命令行传递的参数和加载的全局配置参数，构建PackedProgram对象。PackageProgram对象内包含了Jar包、运行时提交的lib包列表、程序入口函数、savepoint相关配置信息；跟进buildProgram方法内，才此处可以发现，目前Flink主要支持两种类型的执行程序，一种的是python语言，一种是jar包（java or scala)，我们一般通过scala或者java进行Flink程序的开发，所以此处我们重点关注针对jar包的处理逻辑。

在runProgram内，有两个方法，可以重点关注一下，一个是deployJobCluster函数，另外一个是executeProgram;如果程序指定了clusterId,并且运行在Detach模式下，执行deployJobCluster函数；ClusterSpecification有多重实现，分别是Standalone模式、Yarn模式、Dummy模式（主要用于测试）；Flink在我们公司是借助Yarn去做资源的管理，所以我们主要关注Yarn模式下，任务的提交过程；

进入到YarnClusterDescriptor类内，重点关注deployIntenal函数，在这个函数内，程序会去校验用户是否有访问该集群的权限（如果集群开启了kerberos认证模式），指定的队列是否存在、集群当前剩余资源情况是否满足任务提交所申请的资源数等前置条件；校验通过以后，启动startAppMaster，并把AM的host和port，记录到配置项内，方便后期通过RPC接口，获取任务的运行情况；

如果程序没有指定ClusterId或者运行在Atttached模式下,进入executeProgram分支，值得注意的是，这个函数会一直阻塞到任务完成，相关代码如下：

在客户的代码内，以Flink 的streaming 自带的wordcount为例子，在初始化时，需要获取ExecutionEnvironment，调用情况如下：

在创建了运行时所需要的上下文环境后，在用户程序内，通过条用env.execute(jobName)进行提交；

在生成了streamGraph以后（此处忽略如何通过transformations转换成streamGraph）,调用StreamContextEnvironment类内的execute方法，通过判别是Detached模式，还是Attached模式，执行ClientUtils内的对应方法，如下所示：

Flink streaming 任务的提交任务大致如下，后期会探讨一下flink内，如何去保证内部数据流的exactly once的。