深入理解 Flink（六）Flink Job 提交和 Flink Graph 详解

1、获取执行环境对象
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
2、通过执行环境对象，注册数据源 Source，得到数据抽象
DataStream ds = env.socketTextStream(...)
3、调用数据抽象的各种Transformation执行逻辑计算
DataStream resultDS = ds.flatMap(...).keyBy(...).sum(...);
4、将各种Transformation执行完毕之后得到的计算结果数据抽象注册 Sink
resultDS.addSink(...)
5、提交Job执行
env.execute(...)

# Submission to an already running Flink on YARN cluster
./bin/flink run --target yarn-session
# Submission spinning up a Flink on YARN cluster in Per-Job Mode
./bin/flink run --target yarn-per-job
# Submission spinning up Flink on YARN cluster in Application Mode
./bin/flink run-application --target yarn-application

1、提供了 readTextFile(), socketTextStream(), createInput(), addSource() 等方法去对接数据源。
2、提供了 setParallelism() 设置应用程序的并行度。
3、StreamExecutionEnvironment 管理了 ExecutionConfig 对象，该对象负责 Job 执行的一些行为配置管理。还管理了 Configuration 管理一些其他的配置。这个所谓的其他配置，还包含了 Checkpoint 的配置，这个 chekcpoint 的配置参数，会单独解析出来，存储在 CheckpontConfig 中
4、StreamExecutionEnvironment 管理了一个 List> transformations 成员变量，该成员变量，主要用于保存 Job 的各种算子转化得到的 Transformation，把这些 Transformation 按照逻辑拼接起来，就能得到 StreamGragh, 注意转换顺序：
UserFunction ==> StreamOperator ==> Transformation ==> StreamNode
5、StreamExecutionEnvironment 提供了 execute() 方法主要用于提交 Job 执行。该方法接收的参数就是：StreamGraph

// 入口
StreamGraph streamGraph = getStreamGraph(jobName, true){
    // 通过 StreamGraphGenerator 来生成 StreamGraph
    StreamGraph streamGraph = getStreamGraphGenerator().setJobName(jobName).generate(){
        streamGraph = new StreamGraph(....)
        for(Transformation> transformation : transformations) {
            transform(transformation);
        }
    }
}

transform(transformation){
    // 先递归处理该 Transformation 的输入
    Collection inputIds = transform(transform.getInput());
    // 将 Transformation 变成 Operator 设置到 StreamGraph 中，其实就是添加 StreamNode
    streamGraph.addOperator(....);
    // 设置该 StreamNode 的并行度
    streamGraph.setParallelism(transform.getId(), parallelism);
    // 设置该 StreamNode 的入边 SreamEdge
    for(Integer inputId : inputIds) {
        streamGraph.addEdge(inputId, transform.getId(), 0);
        // 内部实现
        // 构建 StreamNode 之间的 边（StreamEdge） 对象
        StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, ...){
            // TODO_MA 注释： 给 上游 StreamNode 设置 出边
            getStreamNode(edge.getSourceId()).addOutEdge(edge);
        }
        // TODO_MA 注释： 给 下游 StreamNode 设置 入边
        getStreamNode(edge.getTargetId()).addInEdge(edge);
    }
}

// 入口
JobClient jobClient = executeAsync(streamGraph){
    // 执行一个 Sream服务器托管网Graph
    executorFactory.getExecutor(configuration).execute(streamGraph, configuration){
        // 第一件事：由 StreamGraph 生成 JobGragh
        JobGraph jobGraph = PipelineExecutorUtils.getJobGraph(pipeline, configuration);
        // 第二件事：通过 RestClusterClient 提交 JobGraph 到Flink集群
        clusterClient.submitJob(jobGraph)
    }
}

// 通过 RestClusterClient 来提交 JobGraph
RestClusterClient.submitJob(JobGraph jobGraph){
    // 继续提交
    RestClusterClient.sendRetriableRequest(){
        // 通过 RestClient 提交
        RestClient.sendRequest(webMonitorHost, webMonitorPort, ...){
            // 继续提交
            RestClient.submitRequest(targetAddress,targetPort,httpRequest,responseType)
        }
    }
}

01、用户根据 Flink 应用程序的编写套路，写好应用程序，打成 jar 包，通过 flink run 的命令来执行提交
02、这个命令的底层，其实是执行： CliFrontend 组件来执行提交
03、这个 CliFrontend 的内部，会通过反射的技术，来转交执行到用户自定义应用程序的 main()
04、先获取 StreamExecutionEnvironment 执行环境对象实例
05、执行算子：其实就是从 算子 ---> function ---> StreamOperator ---> Transformation
06、执行 StreamExecutionEnvironment 的 executor 方法来执行提交
07、首先遍历 StreamExecutionEnvironment 的 transformations 这个 list 来生成 StreamGraph,之后会继续被构建成 JobGraph
08、具体的内部的提交是通过 RestClusterClient 来执行提交
09、在通过 RestClusterClient 提交之前，其实还会做一件事：把 SreamGraph 变成 JobGraph，也还会先把 JobGraph 持久化成为一个磁盘文件
10、在这个 RestClusterClient 的内部，其实是通过 RestClient 来提交
11、RestClient 其实在初始化的时候，就初始化了一个 Netty 客户端
12、通过封装一个 HttpRequest 对象，包含了需要提交的 JobGraph 文件和 Jar 包等，通过 Netty 客户端链接服务端，发送请求对象到服务端
13、Flink 主节点 JobManager 负责处理这个请求的是 WebMonitorEndpoint 中的 Netty 服务端，接收到 rest 请求会调用 Router 执行 route 处理，找到对应的 Handler 执行处理。提交 Job 对应的 Handler 是 JobSubmitHandler

// JobManager 服务端处理入口
JobSubmitHandler.handleRequest(){
    // 恢复得到 JobGraph
    JobGraph jobGraph = loadJobGraph(requestBody, nameToFile);
    // 通过 Dispatcher 提交 JobGraph
    Dispatcher.submitJob(jobGraph, timeout);
}

// 启动 JobMaster
jobMaster.start(){
    JobMaster.onStart(){
        startJobExecution(){
            // 第一件大事：启动 JobMaster 必要的一些工作
            startJobMasterServices(){
                // 第一件事： 启动心跳机制
                this.taskManagerHeartbeatManager = createTaskManagerHeartbeatManager(heartbeatServices);
                this.resourceManagerHeartbeatManager = createResourceManagerHeartbeatManager(heartbeatServices);
                // 第二件事： 启动 SlotPoolImpl
                slotPoolService.start(getFencingToken(), getAddress(), getMainThreadExecutor());
                // 第三件事： 从 ZK 获取 ResourceManager 的地址
                // 这儿就是 JobMaster 向 ResourceManager 执行注册的入口
                resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());
            }
            // 第二件大事：开始调度执行
            startScheduling();
        }
    }
}

ResourceManager.registerJobManager(){
    // ResourceManager 关于 JobMaster 的注册内部实现，重要的事情做了四件
    registerJobMasterInternal(jobMasterGateway, jobId, ....){
        // TODO_MA 马中华 注释： 生成 JobMaster 注册对象
        JobManagerRegistration jobManagerRegistration = new JobManagerRegistration(jobId, jobManagerResourceId, ....);
        // TODO_MA 马中华 注释： 完成注册
        jobManagerRegistrations.put(jobId, jobManagerRegistration);
        jmResourceIdRe服务器托管网gistrations.put(jobManagerResourceId, jobManagerRegistration);
        // TODO_MA 马中华 注释： 加入心跳管理
        jobManagerHeartbeatManager.monitorTarget(jobManagerResourceId, new HeartbeatTarget() {});
        // TODO_MA 马中华 注释： 返回 JobMaster 注册成功
        return new JobMasterRegistrationSuccess(getFencingToken(), resourceId);
    }
}

public class StreamNode {
    private final int id;
    private int parallelism;
    private List inEdges = new ArrayList();
    private List outEdges = new ArrayList();
    private final Class extends AbstractInvokable> jobVertexClass;
}

public class StreamEdge implements Serializable {
    private final String edgeId;
    private final int sourceId;
    private final int targetId;
}

1、JobVertex：经过优化后符合条件的多个 StreamNode 可能会 chain 在一起生成一个 JobVertex，即一个JobVertex 包含一个或多个 operator，JobVertex 的输入是 JobEdge，输出是 IntermediateDataSet。
2、IntermediateDataSet：表示 JobVertex 的输出，即经过 operator 处理产生的数据集。producer 是 JobVertex，consumer 是 JobEdge。
3、JobEdge：代表了 job graph 中的一条数据传输通道。source 是 IntermediateDataSet，target 是 JobVertex。即数据通过 JobEdge 由 IntermediateDataSet 传递给目标 JobVertex。

// 1、下游节点的入度为1 （也就是说下游节点没有来自其他节点的输入）
downStreamVertex.getInEdges().size() == 1;
// 2、上下游节点都在同一个 slot group 中
upStreamVertex.isSameSlotSharingGroup(downStreamVertex);
// 3、前后算子不为空
!(downStreamOperator == null || upStreamOperator == null);
// 4、上游节点的 chain 策略为 ALWAYS 或 HEAD（只能与下游链接，不能与上游链接，Source 默认是 HEAD）
!upStreamOperator.getChainingStrategy() == ChainingStrategy.NEVER;
// 5、下游节点的 chain 策略为 ALWAYS（可以与上下游链接，map、flatmap、filter 等默认是 ALWAYS）
!downStreamOperator.getChainingStrategy() != ChainingStrategy.ALWAYS;
// 6、两个节点间物理分区逻辑是 ForwardPartitioner
(edge.getPartitioner() instanceof ForwardPartitioner);
// 7、两个算子间的 shuffle 方式不等于批处理模式
edge.getShuffleMode() != ShuffleMode.BATCH;
// 8、上下游的并行度一致
upStreamVertex.getParallelism() == downStreamVertex.getParallelism();
// 9、用户没有禁用 chain
streamGraph.isChainingEnabled();

1、在 connect 之间，调用的 createChain() 就是先执行优化，然后再生成 JobVertex
2、然后 调用 connect 之后，是为了组织关系
    1、先生成 IntermediateDataSet 和 JobEdge
    2、把 IntermediateDataSet 和 当前 JobVertex 设置为 JobEdge 的 source 和 target
    3、把 JobEdge 设置为这个 IntermediateDataSet 的消费者

public class JobVertex implements java.io.Serializable {
    private final JobVertexID id;
    private final ArrayList results = new ArrayList();
    private final ArrayList inputs = new ArrayList();
    private int parallelism = ExecutionConfig.PARALLELISM_DEFAULT;
    private String invokableClassName;
}

public class IntermediateDataSet implements java.io.Serializable {
    private final IntermediateDataSetID id;
    private final JobVertex producer;
    private final List consumers = new ArrayList();
}

public class JobEdge implements java.io.Serializable {
    private final JobVertex target;
    private IntermediateDataSet source;
    private IntermediateDataSetID sourceId;
}