Flink 1.17教程：算子链Operator Chain

你的名字 2023-10-15 14:39 41阅读 0赞

## 算子链Operator Chain ##

在 Apache Flink 中，算子链（Operator Chaining）是将多个操作符（算子）连接在一起形成一个链式结构的优化技术。算子链的作用是将多个操作符合并为一个单一的任务单元，以减少通信开销、提高执行效率和减少资源占用。

通俗来说，算子链的作用可以比喻为将多个操作合并成一个整体，就像是把多个小任务捆绑在一起，让它们作为一个整体来执行，而不是分散地执行。

应用场景：

1.  减少通信开销：在一个算子链中，多个操作符可以在同一个线程中执行，避免了数据的序列化和网络通信开销。这对于那些需要频繁传输数据的操作符来说特别有用，可以大大减少数据在不同操作符之间的传输成本。
2.  提高执行效率：算子链可以减少任务切换的开销。在一个算子链中，多个操作符可以在同一个线程中执行，避免了线程之间的切换，并且可以通过内存数据传递来提高执行效率。
3.  减少资源占用：算子链可以减少并行任务的数量，从而减少了需要的计算资源。如果多个操作符在一个算子链中执行，它们可以共享同一个线程或任务资源，这样可以有效地降低整体资源的占用。

总的来说，算子链可以在 Apache Flink 中提供优化的执行方式，减少通信开销、提高执行效率和减少资源占用。它适用于需要频繁传输数据、有依赖关系的操作符，以及需要提高整体执行效率和减少资源消耗的场景。

### 合并算子链 ###

在Flink中，并行度相同的一对一（one to one）算子操作，可以直接链接在一起形成一个“大”的任务（task），这样原来的算子就成为了真正任务里的一部分，如下图所示。每个task会被一个线程执行。这样的技术被称为“算子链”（Operator Chain）。

![img][]

## 算子链演示 ##

1、算子之间的传输关系：  
一对一  
重分区  
2、算子 串在一起的条件：  
1） 一对一  
2） 并行度相同  
3、关于算子链的api：  
1）全局禁用算子链：`env.disableOperatorChaining();`  
2）某个算子不参与链化： `算子A.disableChaining()`, 算子A不会与 前面 和 后面的算子 串在一起  
3）从某个算子开启新链条： `算子A.startNewChain()`， 算子A不与 前面串在一起，从A开始正常链化

package com.atguigu.wc;
     
    import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
     
    /**
     * TODO DataStream实现Wordcount：读socket（无界流）
     *
     * @author
     * @version 1.0
     */
    public class OperatorChainDemo {
        
        public static void main(String[] args) throws Exception {
        
            // TODO 1.创建执行环境
    		// StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // IDEA运行时，也可以看到webui，一般用于本地测试
            // 需要引入一个依赖 flink-runtime-web
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
     
            // 在idea运行，不指定并行度，默认就是 电脑的 线程数
            env.setParallelism(1);
     
            // 全局禁用 算子链
    		//env.disableOperatorChaining();
     
            // TODO 2.读取数据：socket
            DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 7777);
     
            // TODO 3.处理数据: 切换、转换、分组、聚合
            SingleOutputStreamOperator<Tuple2<String,Integer>> sum = socketDS
    				//.disableChaining()
                    .flatMap(
                            (String value, Collector<String> out) -> {
        
                                String[] words = value.split(" ");
                                for (String word : words) {
        
                                    out.collect(word);
                                }
                            }
                    )
                    .startNewChain()
    				//.disableChaining()
                    .returns(Types.STRING)
                    .map(word -> Tuple2.of(word, 1))
                    .returns(Types.TUPLE(Types.STRING,Types.INT))
                    .keyBy(value -> value.f0)
                    .sum(1);
     
            // TODO 4.输出
            sum.print();
     
            // TODO 5.执行
            env.execute();
        }
    }
     
    /**
     1、算子之间的传输关系：
         一对一
         重分区
     2、算子 串在一起的条件：
        1） 一对一
        2） 并行度相同
     3、关于算子链的api：
        1）全局禁用算子链：env.disableOperatorChaining();
        2）某个算子不参与链化：  算子A.disableChaining(),  算子A不会与 前面 和 后面的算子 串在一起
        3）从某个算子开启新链条：  算子A.startNewChain()， 算子A不与 前面串在一起，从A开始正常链化
     */

[img]: https://img-blog.csdnimg.cn/d7248b5fa87140d58b02488a33cd40f1.png