Flink 原理与实现:Operator Chain原理

Flink原理与实现系列文章 :

Flink 原理与实现:架构和拓扑概览
Flink 原理与实现:如何生成 StreamGraph
Flink 原理与实现:如何生成 JobGraph
Flink原理与实现:如何生成ExecutionGraph及物理执行图

Flink的逻辑/执行计划优化,有一个很大的特点就是,会将多个operator,串在一起作为一个operator chain来执行。关于operator chain,在 Flink 原理与实现:理解 Flink 中的计算资源 中已经有了初步的介绍,在阅读本文之前,建议先阅读上文。
本文将从源码上进一步分析,探究operator chain内部是如何实现的。

OperatorChain是在StreamTask的invoke方法中被创建的:

     // ...
        operatorChain = new OperatorChain<>(this);
        headOperator = operatorChain.getHeadOperator();
        // ...

Flink原理与实现:如何生成ExecutionGraph及物理执行图中提到,StreamTask是真正的执行task中的invokable operator(的基类),因此所有的task都会创建OperatorChain这个对象。只是在执行的时候,如果一个operator无法被chain起来,那它就只有headOperator,chain里就没有其他operator了。

OperatorChain构造函数:

            List<StreamOperator<?>> allOps = new ArrayList<>(chainedConfigs.size());
            this.chainEntryPoint = createOutputCollector(containingTask, configuration,
                    chainedConfigs, userCodeClassloader, streamOutputMap, allOps);

            if (headOperator != null) {
                headOperator.setup(containingTask, configuration, getChainEntryPoint());
            }

            // add head operator to end of chain
            allOps.add(headOperator);

这里headerOperator.setup方法第三个参数为Output,相当于把chainEntryPoint作为output传入head operator。setup方法一路调用,直到基类AbstractStreamOperator,可以看到:

    this.output = new CountingOutput(output,((OperatorMetricGroup)this.metrics).getIOMetricGroup().getNumRecordsOutCounter());

即对output封装成了AbstractStreamOperator.CountingOutput,主要是为了统计metrics信息。

而output自身在operator chain中,是一个CopyingChainingOutput,或者ChainingOutput(根据是否配置了reuse objects)。

这里的headOperator即为operator chain中第一个operator,在这里即为StreamGroupedReduce。
它在执行processElement的时候,如果有调用output.collect,则会调用CountingOutput。它的collect方法很简单:

        @Override
        public void collect(StreamRecord<OUT> record) {
            numRecordsOut.inc();
            output.collect(record);
        }

即更新metrics和调用ChainingOutput.collect方法,看看这个方法:

        @Override
        public void collect(StreamRecord<T> record) {
            try {
                numRecordsIn.inc();
                StreamRecord<T> copy = record.copy(serializer.copy(record.getValue()));
                operator.setKeyContextElement1(copy);
                operator.processElement(copy);
            }
            catch (Exception e) {
                throw new RuntimeException("Could not forward element to next operator", e);
            }
        }

这里的operator是chainedOperator,即除了headOperator之外,剩余的operators的chain。
调用这个operator.processElement,就会像上面一样,循环调用operator chain里的所有operator,一直到chain end。

以word count为例,应用代码如下:

    // ...
    DataStream<String> text = env.fromElements(WordCountData.WORDS);
    DataStream<Tuple2<String, Integer>> counts =
            // split up the lines in pairs (2-tuples) containing: (word,1)
            text.flatMap(new Tokenizer())
                    // group by the tuple field "0" and sum up tuple field "1"
                    .keyBy(0).sum(1).filter(new FilterFunction<Tuple2<String, Integer>>() {
                @Override
                public boolean filter(Tuple2<String, Integer> value) throws Exception {
                    return value.f1 > 1;
                }
            });
      env.execute("Streaming WordCount");

它实际上形成了以下的调用链:

StreamGroupedReduce.processElement
--> CountingOutput.collect
--> CopyChainingOutput.collect
    --> StreamFilter.processElement
    --> CountingOutput.collect
    --> CopyChainingOutput.collect
        --> StreamSink.processElement
        --> CountingOutput.collect
        --> BroadcastingOutputCollector.collect

下面会解析如何生成这个调用链。我们返回到OperatorChain的构造函数中,看一下这行代码:

            this.chainEntryPoint = createOutputCollector(containingTask, configuration,
                    chainedConfigs, userCodeClassloader, streamOutputMap, allOps);

到底做了什么。

这个方法的重要代码如下:

        // 遍历当前operatorConfig的输出边
        for (StreamEdge outputEdge : operatorConfig.getChainedOutputs(userCodeClassloader)) {
           // 下游operator id
            int outputId = outputEdge.getTargetId();
            // 得到下游operator的stream config
            StreamConfig chainedOpConfig = chainedConfigs.get(outputId);
        
        // 根据下游operator的stream config,创建chained operator
            Output<StreamRecord<T>> output = createChainedOperator(
                    containingTask,
                    chainedOpConfig,
                    chainedConfigs,
                    userCodeClassloader,
                    streamOutputs,
                    allOperators);
            allOutputs.add(new Tuple2<>(output, outputEdge));
        }

再看下createChainedOperator方法:

     // 第一行就递归调用了createOutputCollector方法,创建当前operator下游operator的collector
        Output<StreamRecord<OUT>> output = createOutputCollector(
                containingTask, operatorConfig, chainedConfigs, userCodeClassloader, streamOutputs, allOperators);

     // setup当前operator,其实是把下游operator的collector作为当前operator的output
     // 这样当前operator调用collect的时候,就会调用下游operator的方法。
        OneInputStreamOperator<IN, OUT> chainedOperator = operatorConfig.getStreamOperator(userCodeClassloader);
        chainedOperator.setup(containingTask, operatorConfig, output);

        allOperators.add(chainedOperator);

      // 根据是否reuse object,创建ChainingOutput或者CopyingChainingOutput
        if (containingTask.getExecutionConfig().isObjectReuseEnabled()) {
            return new ChainingOutput<>(chainedOperator);
        }
        else {
            TypeSerializer<IN> inSerializer = operatorConfig.getTypeSerializerIn1(userCodeClassloader);
            return new CopyingChainingOutput<>(chainedOperator, inSerializer);
        }

由于这个过程是递归的,所以chained operators实际上是从下游往上游去反向一个个创建和setup的。以word count为例,chained operators为:StreamGroupedReduce - StreamFilter - StreamSink,而实际初始化顺序则相反:StreamSink - StreamFilter - StreamGroupedReduce。

在OperatorChain类中,headOperator为StreamGroupedReduce。createOutputCollector的调用过程如下:

createOutputCollector(operatorConfig=<StreamGroupedReduce config>, ...)
 --> chainedOpConfig = <StreamFilter config>
 --> createChainedOperator(chainedOpConfig=<StreamFilter config>)
    --> createOutputCollector(<StreamFilter config>)
    --> chainedOpConfig = <StreamSink config>
        --> createChainedOperator(<StreamSink config>)
            --> createOutputCollector(<StreamSink config>)
            --> chainedOpConfig = null, 返回BroadcastingOutputCollector
            --> StreamSink.setup(<output=BroadcastingOutputCollector>)
            --> return CopyingChainingOutput
    --> output = CopyingChainingOutput
    --> StreamFilter.setup(<output=CopyingChainingOutput>)
    --> return CopyingChainingOutput
--> output = CopyingChainingOutput
--> headOperator.setup(<output=CopyingChainingOutput>)            

最后我们来看一下,如果operator chain中只有一个operator的情况,它生成了什么。
在word count的例子中,在StreamSource之后的flatMap,就是这种情况,它不能跟后面的操作chain在一起。

首先OperatorChain构造函数中的chainedConfigs会为空,因为下游没有跟它chain在一起的operator。接下来看下它的chainEntryPoint

createOutputCollector方法中,由于没有chained outputs,因此会直接返回RecordWriterOutput,即headOperator的output就直接交给record writer输出了。

上一篇:Flink 原理与实现:架构和拓扑概览


下一篇:apache工作模式介绍