一個Flink作業(yè)，從client提交到真正的執(zhí)行，其 Graph 的轉(zhuǎn)換會經(jīng)過下面三個階段（第四個階段是作業(yè)真正執(zhí)行時的狀態(tài)，都是以 task 的形式在 TM 中運行）：

StreamGraph：根據(jù)編寫的代碼生成最初的 Graph，它表示最初的拓?fù)浣Y(jié)構(gòu)；
JobGraph：這里會對前面生成的 Graph，做一些優(yōu)化操作（比如: operator chain 等），最后會提交給 JobManager；
ExecutionGraph：JobManager 根據(jù) JobGraph 生成 ExecutionGraph，是 Flink 調(diào)度時依賴的核心數(shù)據(jù)結(jié)構(gòu)；
物理執(zhí)行圖：JobManager 根據(jù)生成的 ExecutionGraph 對 Job 進(jìn)行調(diào)度后，在各個 TM 上部署 Task 后形成的一張?zhí)摂M圖。
首先要熟悉以下概念：DataStream、Transformation、StreamOperator、Function

DataStream

image.png

DateStream實際上就是對相同類型的數(shù)據(jù)流操作進(jìn)行的封裝，主要是通過Transformations將數(shù)據(jù)流轉(zhuǎn)換成另一個流，常用的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()，調(diào)用了StreamExecutionEnvironment的addSource()方法，會講SourceFunction封裝到StreamSource中，之后會講StreamSource封裝到SourceTransformation中。所以最終執(zhí)行業(yè)務(wù)核心邏輯的是各種Function，經(jīng)過各種轉(zhuǎn)換會生成對應(yīng)的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，最終都返回?fù)碛挟?dāng)前算子和環(huán)境的DataStream對象，OneInputTransformation算法擁有當(dāng)前算子的上游算子對象。整個封裝過程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
                        * 記錄當(dāng)前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，它主要的實現(xiàn)類如下：
AbstractUdfStreamOperator：會封裝一個 Function，真正的操作是在 Function 中的實現(xiàn)。
OneInputStreamOperator：如果這個 Operator 只有一個輸入，實現(xiàn)這個接口即可，processElement() 方法需要自己去實現(xiàn)，主要做業(yè)務(wù)邏輯的處理；
TwoInputStreamOperator：如果這個 Operator 是一個二元操作符，是對兩個流的處理，比如：雙流 join，那么實現(xiàn)這個接口即可，自己去實現(xiàn) processElement1() 和 processElement2() 方法。

image.png

Function

Function 是 Transformation 最底層的封裝，用戶真正的處理邏輯是在這個里面實現(xiàn)的，包括前面示例中實現(xiàn)的 FlatMapFunction 對象等。

image.png

以上熟悉各個概念之后，以及各個類之間的關(guān)系，對于后邊StreamGraph的生成過程有極大的幫助。