Job的提交流程（一）

一个Flink作业，从client提交到真正的执行，其 Graph 的转换会经过下面三个阶段（第四个阶段是作业真正执行时的状态，都是以 task 的形式在 TM 中运行）：

StreamGraph：根据编写的代码生成最初的 Graph，它表示最初的拓扑结构；
JobGraph：这里会对前面生成的 Graph，做一些优化操作（比如: operator chain 等），最后会提交给 JobManager；
ExecutionGraph：JobManager 根据 JobGraph 生成 ExecutionGraph，是 Flink 调度时依赖的核心数据结构；
物理执行图：JobManager 根据生成的 ExecutionGraph 对 Job 进行调度后，在各个 TM 上部署 Task 后形成的一张虚拟图。
首先要熟悉以下概念：DataStream、Transformation、StreamOperator、Function

DataStream

image.png

DateStream实际上就是对相同类型的数据流操作进行的封装，主要是通过Transformations将数据流转换成另一个流，常用的api:
keyBy()、join()、union()、map()、filter()、flatMap()等。

Transformation

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

       // get input data by connecting to the socket
       DataStream<String> text = env.socketTextStream(hostname, port, "\n");

       // parse the data, group it, window it, and aggregate the counts
       DataStream<WordWithCount> windowCounts =
               text.flatMap(
                               new FlatMapFunction<String, WordWithCount>() {
                                   @Override
                                   public void flatMap(
                                           String value, Collector<WordWithCount> out) {
                                       for (String word : value.split("\\s")) {
                                           out.collect(new WordWithCount(word, 1L));
                                       }
                                   }
                               })
                       .keyBy(value -> value.word)
                       .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                       .reduce(
                               new ReduceFunction<WordWithCount>() {
                                   @Override
                                   public WordWithCount reduce(WordWithCount a, WordWithCount b) {
                                       return new WordWithCount(a.word, a.count + b.count);
                                   }
                               });

       // print the results with a single thread, rather than in parallel
       windowCounts.print().setParallelism(1);

       env.execute("Socket Window WordCount");

首先看env.socketTextStream()，调用了StreamExecutionEnvironment的addSource()方法，会讲SourceFunction封装到StreamSource中，之后会讲StreamSource封装到SourceTransformation中。所以最终执行业务核心逻辑的是各种Function，经过各种转换会生成对应的Transformation。

private <OUT> DataStreamSource<OUT> addSource(
            final SourceFunction<OUT> function,
            final String sourceName,
            @Nullable final TypeInformation<OUT> typeInfo,
            final Boundedness boundedness) {
        checkNotNull(function);
        checkNotNull(sourceName);
        checkNotNull(boundedness);

        TypeInformation<OUT> resolvedTypeInfo =
                getTypeInfo(function, sourceName, SourceFunction.class, typeInfo);

        boolean isParallel = function instanceof ParallelSourceFunction;

        clean(function);

        final StreamSource<OUT, ?> sourceOperator = new StreamSource<>(function);
      //DataStreamSource 其实是SingleOutputStreamOperator的子类
        return new DataStreamSource<>(
                this, resolvedTypeInfo, sourceOperator, isParallel, sourceName, boundedness);
    }

再看flatMap()操作，其实和上述类似，会将Function封装成StreamOperator，再封装成Transformation，最终都返回拥有当前算子和环境的DataStream对象，OneInputTransformation算法拥有当前算子的上游算子对象。整个封装过程Function->StreamOperator->Transformation

image.png

   public <R> SingleOutputStreamOperator<R> flatMap(
            FlatMapFunction<T, R> flatMapper, TypeInformation<R> outputType) {
        return transform("Flat Map", outputType, new StreamFlatMap<>(clean(flatMapper)));
    }

protected <R> SingleOutputStreamOperator<R> doTransform(
            String operatorName,
            TypeInformation<R> outTypeInfo,
            StreamOperatorFactory<R> operatorFactory) {

        // read the output type of the input Transform to coax out errors about MissingTypeInfo
        transformation.getOutputType();

        OneInputTransformation<T, R> resultTransform =
                new OneInputTransformation<>(
                        /**
                        * han_pf
                        * 记录当前transformation的输入transformation
                        */
                        this.transformation,
                        operatorName,
                        operatorFactory,
                        outTypeInfo,
                        environment.getParallelism());

        @SuppressWarnings({"unchecked", "rawtypes"})
        SingleOutputStreamOperator<R> returnStream =
                new SingleOutputStreamOperator(environment, resultTransform);
        /**
        * han_pf
        * 存储所有的stransformation到env中。
        */
        getExecutionEnvironment().addOperator(resultTransform);

        return returnStream;
    }

StreamOperator

Operator 基类的是 StreamOperator，它表示的是对 Stream 的一个 operation，它主要的实现类如下：
AbstractUdfStreamOperator：会封装一个 Function，真正的操作是在 Function 中的实现。
OneInputStreamOperator：如果这个 Operator 只有一个输入，实现这个接口即可，processElement() 方法需要自己去实现，主要做业务逻辑的处理；
TwoInputStreamOperator：如果这个 Operator 是一个二元操作符，是对两个流的处理，比如：双流 join，那么实现这个接口即可，自己去实现 processElement1() 和 processElement2() 方法。

image.png

Function

Function 是 Transformation 最底层的封装，用户真正的处理逻辑是在这个里面实现的，包括前面示例中实现的 FlatMapFunction 对象等。

image.png

以上熟悉各个概念之后，以及各个类之间的关系，对于后边StreamGraph的生成过程有极大的帮助。