开发者社区> codyinnowhere> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

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 版与自身环境打通|学习笔记
快速学习 如何将实时计算 Flink 版与自身环境打通
36 0
贝壳基于 Flink 的实时计算演进之路
贝壳找房在实时计算之路上的平台建设以及实时数仓应用。
6037 0
【必看】如何正确使用实时计算 Flink 版?
本篇文章将从实时计算 Flink 版产品功能、产品架构、产品模式、产品优势、产品应用场景等全面呈现,同时还汇总了实时计算 Flink 版学习资料!更有特惠独享活动限时参与!
3080 0
如何开通实时计算 Flink 版
本文由阿里巴巴产品专家赵开兴分享,主要介绍实时计算 Flink的开通、功能及使用,并通过多个演示示例进行详细demo演示。内容主要分享以下几部分:1、Blink独享集群使用介绍;2、Flink 全托管集群的使用
1792 0
【行业应用】阿里云实时计算 Flink 版游戏行业解决方案
游戏作为新兴崛起的娱乐产业,当下发展得如火如荼,其吸金能力和趣味性也吸引更多企业与人才投入其中。游戏行业公司主要分为发行和制作两类,游戏的类型可细分为手游、页游和端游三种,随着移动端设备的更新发展以及 5G 时代的降临,手游将迎来绝佳的发展时机。
2678 0
开源 Flink + 实时计算 Flink 版训练营学习资料汇总
Apache Flink 是一个开源的分布式大数据处理引擎, 可对有限数据流和无限数据流进行有状态计算。实时计算 Flink版是阿里云提供的基于 Apache Flink 构建的企业级、高性能实时大数据处理系统,由Apache Flink创始团队官方出品。
4981 0
实时计算 Flink 版 最佳实践
实时计算 Flink 版 最佳实践目录
9138 0
ReentrantLock实现原理
同步状态 /** * The synchronization state. */ private volatile int state; /** * Returns the current value of synchronization state.
883 0
4
文章
10
问答
文章排行榜
最热
最新
相关电子书
更多
低代码开发师(初级)实战教程
立即下载
阿里巴巴DevOps 最佳实践手册
立即下载
冬季实战营第三期:MySQL数据库进阶实战
立即下载