Flink DataStream 算子梳理

约定不等于承诺〃 2022-11-01 04:23 363阅读 0赞

### . ###

*  一 .前言
 *   *  1.1. Flink算子分类
     *  1.2. 数据类型转换关系
 *  二 .算子清单
 *   *  2.1. Map
     *  2.2. FlatMap
     *  2.3. Filter
     *  2.4. KeyBy
     *  2.5. Reduce
     *  2.6. Fold \[废弃\]
     *  2.7. Aggregation
     *  2.8. Window
     *  2.9. WindowAll
     *  2.10. Window Apply
     *  2.11. Window reduce
     *  2.12. Window Fold \[废弃\]
     *  2.13. Window Aggregation
     *  2.14. Window Join
     *  2.15.Union
     *  2.16. Interval Join
     *  2.17. WindowCoGroup
     *  2.18. Connect
     *  2.19. CoMap 和 CoFlatMap
     *  2.20. Split \[废弃\]
     *  2.21. Select \[废弃\]
     *  2.22. Iterate
     *  2.23. Extract Timestamps
     *  2.24. Project
 *  三.物理分区
 *   *  3.1. Custom partitioning
     *  3.2. Random partitioning
     *  3.3. Rebalancing (Round-robin partitioning)
     *  3.4. Rescaling
     *  3.5. Broadcasting
 *  四. 算子链和资源组
 *   *  4.1.Start new chain
     *  4.2.Disable chaining
     *  4.3.Set slot sharing group

# 一 .前言 #

Apache Flink 是一个分布式大数据处理引擎，可对有限数据流和无限数据流进行有状态或无状态的计算，能够部署在各种集群环境，对各种规模大小的数据进行快速计算。  
![Flink运行时架构][Flink]

DataStream API 使用了Fluent风格处理数据, 开发的时候其实是在编写一个DataStream转换的过程. 在这里,我们对Flink算子做一个梳理.

## 1.1. Flink算子分类 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

Flink DataStream API 的核心，就是代表流数据的 DataStream 对象。

整个计算逻辑图的构建就是围绕调用 DataStream 对象上的不同操作产生新的 DataStream 对象展开的。

整体来说，DataStream 上的操作可以分为四类。

*  第一类 : 对单条记录的操作，比如筛除掉不符合要求的记录（Filter 操作），或者将每条记录都做一个转换（Map 操作）。
 *  第二类 : 对多条记录的操作。比如说统计一个小时内的订单总成交量，就需要将一个小时内的所有订单记录的成交量加到一起。为了支持这种类型的操作，就得通过 Window 将需要的记录关联到一起进行处理。
 *  第三类 : 对多个流进行操作并转换为单个流。例如，多个流可以通过 Union、Join 或 Connect 等操作合到一起。这些操作合并的逻辑不同，但是它们最终都会产生了一个新的统一的流，从而可以进行一些跨流的操作。
 *  第四类 : 分组操作 keyBy 。

## 1.2. 数据类型转换关系 ##

为了支持这些不同的流操作，Flink 引入了一组不同的流类型，用来表示某些操作的中间流数据集类型。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

# 二 .算子清单 #

## 2.1. Map ##

DataStream --> DataStream ：是对输入的参数进行转换操作, 输入一个参数经过处理后转换为另外一个参数 .

DataStream<Integer> dataStream = //...
    dataStream.map(new MapFunction<Integer, Integer>() { 
        @Override
        public Integer map(Integer value) throws Exception { 
            return 2 * value;
        }

import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    //这个例子是监听9000 socket端口，对于发送来的数据，以\n为分隔符分割后进行处理，
    //将分割后的每个元素，添加上一个字符串后，打印出来。
    public class MapDemo { 
        private static int index = 1;
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream = env.socketTextStream("localhost", 9000, "\n");
            //3.map操作。
            DataStream<String> result = textStream.map(s -> (index++) + ".您输入的是：" + s);
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    }

## 2.2. FlatMap ##

DataStream --> DataStream：用于拆分操作 , 输入一个参数(集合)，产生0、1或者多个输出参数.

dataStream.flatMap(new FlatMapFunction<String, String>() { 
        @Override
        public void flatMap(String value, Collector<String> out)
            throws Exception { 
            for(String word: value.split(" ")){ 
                out.collect(word);
            }
        }
    });

import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
    
    //这个例子是用Flink监听9000端口，将接受的字符串用\n分割为一个个的元素
    //然后将每个元素拆为一个个的字符，并打印出来
    public class FlatMapDemo { 
        private static int index1 = 1;
        private static int index2 = 1;
    
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream = env.socketTextStream("localhost", 9000, "\n");
            //3.flatMap操作，对每一行字符串进行分割
            DataStream<String> result = textStream.flatMap((String s, Collector<String> collector) -> { 
                for (String str : s.split("")) { 
                    collector.collect(str);
                }
            })
                    //这个地方要注意，在flatMap这种参数里有泛型算子中。
                    //如果用lambda表达式，必须将参数的类型显式地定义出来。
                    //并且要有returns，指定返回的类型
                    //详情可以参考Flink官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/java_lambdas.html
                    .returns(Types.STRING);
    
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    }

## 2.3. Filter ##

DataStream --> DataStream：根据条件过滤数据，并返回为true的元素

dataStream.filter(new FilterFunction<Integer>() { 
        @Override
        public boolean filter(Integer value) throws Exception { 
            return value != 0;
        }
    });

import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class FilterDemo { 
        private static int index = 1;
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream = env.socketTextStream("localhost", 9000, "\n");
            //3.filter操作，筛选非空行。
            DataStream<String> result = textStream.filter(line->!line.trim().equals(""));
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    }

## 2.4. KeyBy ##

DataSteam --> KeyedStream：根据key的散列值进行分区的, 将DataStream中的数据根据指定的Key进行分区。

**注意：以下类型无法作为key**

1.  POJO类，且没有实现hashCode函数
2.  任意形式的数组类型

dataStream.keyBy(value -> value.getSomeKey()) // Key by field "someKey"
    dataStream.keyBy(value -> value.f0) // Key by the first element of a Tuple

import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.api.java.functions.KeySelector;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.time.Time;
    
    import java.util.concurrent.TimeUnit;
    
    //这个例子是每行输入一个单词，以单词为key进行计数
    //每10秒统计一次每个单词的个数
    public class KeyByDemo { 
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream = env.socketTextStream("localhost", 9000, "\n");
            //3.
            DataStream<Tuple2<String, Integer>> result = textStream
                    //map是将每一行单词变为一个tuple2
                    .map(line -> Tuple2.of(line.trim(), 1))
                    //如果要用Lambda表示是，Tuple2是泛型，那就得用returns指定类型。
                    .returns(Types.TUPLE(Types.STRING, Types.INT))
                    //keyBy进行分区，按照第一列，也就是按照单词进行分区
                    .keyBy(new KeySelector<Tuple2<String , Integer>, Integer>() { 
                        @Override
                        public Integer getKey(Tuple2<String, Integer> value) throws Exception { 
                            return value.f1;
                        }
                    })
                    //指定窗口，每10秒个计算一次
                    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                    //计算个数，计算第1列
                    .sum(1);
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    }

## 2.5. Reduce ##

KeyedStream --> DataStream：滚动和并操作，合并当前元素和上一次合并的元素结果。

keyedStream.reduce(new ReduceFunction<Integer>() { 
        @Override
        public Integer reduce(Integer value1, Integer value2)
        throws Exception { 
            return value1 + value2;
        }
    });

package org.apache.flink.streaming.api.test;
    
    import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.time.Time;
    
    import java.util.concurrent.TimeUnit;
    
    //这个例子是对流进行分组，分组后进归并操作。
    //是wordcount的另外一种实现方法
    public class ReduceDemo { 
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream = env.socketTextStream("localhost", 9000, "\n");
            //3.
            DataStream<Tuple2<String, Integer>> result = textStream
                    //map是将每一行单词变为一个tuple2
                    .map(line -> Tuple2.of(line.trim(), 1))
                    //如果要用Lambda表示是，Tuple2是泛型，那就得用returns指定类型。
                    .returns(Types.TUPLE(Types.STRING, Types.INT))
                    //keyBy进行分区，按照第一列，也就是按照单词进行分区
                    .keyBy(0)
                    //指定窗口，每10秒个计算一次
                    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                    //对每一组内的元素进行归并操作，即第一个和第二个归并，结果再与第三个归并...
                    .reduce((Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) -> new Tuple2(t1.f0, t1.f1 + t2.f1));
    
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    }

## 2.6. Fold \[废弃\] ##

KeyedStream --> DataStream：用一个初始的一个值的Reudce，与其每个元素进行滚动合并操作。

package org.apache.flink.streaming.api.test;
    
    import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.time.Time;
    
    import java.util.concurrent.TimeUnit;
    
    public class FoldDemo { 
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream = env.socketTextStream("localhost", 9000, "\n");
            //3.
            DataStream<String> result = textStream
                    //map是将每一行单词变为一个tuple2
                    .map(line -> Tuple2.of(line.trim(), 1))
                    //如果要用Lambda表示是，Tuple2是泛型，那就得用returns指定类型。
                    .returns(Types.TUPLE(Types.STRING, Types.INT))
                    //keyBy进行分区，按照第一列，也就是按照单词进行分区
                    .keyBy(0)
                    //指定窗口，每10秒个计算一次
                    .timeWindow(Time.of(10, TimeUnit.SECONDS))
                    //指定一个开始的值，对每一组内的元素进行归并操作，即第一个和第二个归并，结果再与第三个归并...
                    .fold("结果：",(String current, Tuple2<String, Integer> t2) -> current+t2.f0+",");
    
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    }

## 2.7. Aggregation ##

KeyedStream --> DataStream：分组流数据的滚动聚合操作：min和minBy的区别是min返回的是一个最小值，而minBy返回的是其字段中包含的最小值的元素（同样元原理适用于max和maxBy）

keyedStream.sum(0);
    keyedStream.sum("key");
    keyedStream.min(0);
    keyedStream.min("key");
    keyedStream.max(0);
    keyedStream.max("key");
    keyedStream.minBy(0);
    keyedStream.minBy("key");
    keyedStream.maxBy(0);
    keyedStream.maxBy("key");

## 2.8. Window ##

KeyedStream --> DataStream：windows是在一个分区的KeyedStreams中定义的，windows根据某些特性将每个key的数据按照时间的维度划分为滚动的窗口（例如：在5s内到达的数据）。

dataStream.keyBy(value -> value.f0)
    			.window(TumblingEventTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data

## 2.9. WindowAll ##

DataStream --> AllWindowedStream：全局时间窗, Windows可以在一个常规的DataStream中定义.

注意：这个操作在很多情况下都不是并行操作的，所有的记录都会聚集到一个windowAll操作的任务中

dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data

## 2.10. Window Apply ##

将Window函数应用到窗口上,Window函数将一个窗口的数据作为整体进行处理.  
Window Stream有两种 : 分组后的WindowStream和未分组的AllWindowedStream .  
WindowedStream --> DataStream  
AllWindowedStream --> DataStream：将一个通用的函数作为一个整体传递给window。

windowedStream.apply (new WindowFunction<Tuple2<String,Integer>, Integer, Tuple, Window>() { 
        public void apply (Tuple tuple,
                Window window,
                Iterable<Tuple2<String, Integer>> values,
                Collector<Integer> out) throws Exception { 
            int sum = 0;
            for (value t: values) { 
                sum += t.f1;
            }
            out.collect (new Integer(sum));
        }
    });
    
    // applying an AllWindowFunction on non-keyed window stream
    allWindowedStream.apply (new AllWindowFunction<Tuple2<String,Integer>, Integer, Window>() { 
        public void apply (Window window,
                Iterable<Tuple2<String, Integer>> values,
                Collector<Integer> out) throws Exception { 
            int sum = 0;
            for (value t: values) { 
                sum += t.f1;
            }
            out.collect (new Integer(sum));
        }
    });

## 2.11. Window reduce ##

WindowedStream --> DataStream：给窗口赋予一个reduce的功能，并返回一个reduce的结果。

windowedStream.reduce (new ReduceFunction<Tuple2<String,Integer>>() { 
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception { 
            return new Tuple2<String,Integer>(value1.f0, value1.f1 + value2.f1);
        }
    });

## 2.12. Window Fold \[废弃\] ##

WindowedStream --> DataStream：给窗口赋予一个fold的功能，并返回一个DataStream。

## 2.13. Window Aggregation ##

WindowedStream --> DataStream：对window的元素做聚合操作，min和minBy的区别是min返回的是最小值，而minBy返回的是包含最小值字段的元素。（同样原理适用于max和maxBy）

windowedStream.sum(0);
    windowedStream.sum("key");
    windowedStream.min(0);
    windowedStream.min("key");
    windowedStream.max(0);
    windowedStream.max("key");
    windowedStream.minBy(0);
    windowedStream.minBy("key");
    windowedStream.maxBy(0);
    windowedStream.maxBy("key");

## 2.14. Window Join ##

DataStream，DataStream --> DataStream：根据给定的key和window对两个DataStream做join操作

dataStream.join(otherStream)
        .where(<key selector>).equalTo(<key selector>)
        .window(TumblingEventTimeWindows.of(Time.seconds(3)))
        .apply (new JoinFunction () { ...});

package org.apache.flink.streaming.api.test;
    
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    
    public class WindowJoinDemo { 
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream9000 = env.socketTextStream("localhost", 9000, "\n");
            DataStream<String> textStream9001 = env.socketTextStream("localhost", 9001, "\n");
            //将输入处理一下，变为tuple2
            DataStream<Tuple2<String,String>> mapStream9000=textStream9000
                    .map(new MapFunction<String, Tuple2<String,String>>() { 
                        @Override
                        public Tuple2<String, String> map(String s) throws Exception { 
                            return Tuple2.of(s,"来自9000端口："+s);
                        }
                    });
    
            DataStream<Tuple2<String,String>> mapStream9001=textStream9001
                    .map(new MapFunction<String, Tuple2<String,String>>() { 
                        @Override
                        public Tuple2<String, String> map(String s) throws Exception { 
                            return Tuple2.of(s,"来自9001端口："+s);
                        }
                    });
    
            //3.两个流进行join操作，是inner join，关联上的才能保留下来
            DataStream<String> result =  mapStream9000.join(mapStream9001)
                    //关联条件，以第0列关联（两个source输入的字符串）
                    .where(t1->t1.getField(0)).equalTo(t2->t2.getField(0))
                    //以处理时间，每10秒一个滚动窗口
                    .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                    //关联后输出
                    .apply((t1,t2)->t1.getField(1)+"|"+t2.getField(1))
                    ;
    
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    }

## 2.15.Union ##

DataStream --> DataStream：对两个或两个以上的DataStream做union操作，产生一个包含所有的DataStream元素的新DataStream。  
注意：如果将一个DataStream和自己做union操作，在新的DataStream中，将看到每个元素重复两次

dataStream.union(otherStream1, otherStream2, ...);

package org.apache.flink.streaming.api.test;
    
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    //这个例子是将三个socket端口发送来的数据合并到一个流中
    //可以对这三个流发送来的数据，集中处理。
    public class UnionDemo { 
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream9000 = env.socketTextStream("localhost", 9000, "\n");
            DataStream<String> textStream9001 = env.socketTextStream("localhost", 9001, "\n");
            DataStream<String> textStream9002 = env.socketTextStream("localhost", 9002, "\n");
    
            DataStream<String> mapStream9000=textStream9000.map(s->"来自9000端口："+s);
            DataStream<String> mapStream9001=textStream9001.map(s->"来自9001端口："+s);
            DataStream<String> mapStream9002=textStream9002.map(s->"来自9002端口："+s);
    
            //3.union用来合并两个或者多个流的数据，统一到一个流中
            DataStream<String> result =  mapStream9000.union(mapStream9001,mapStream9002);
    
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    }

## 2.16. Interval Join ##

KeyedStream --> DataStream： 对两个KeyedStream进行Join, 需要指定时间范围和Join时使用的Key .

// this will join the two streams so that
    // key1 == key2 && leftTs - 2 < rightTs < leftTs + 2
    keyedStream.intervalJoin(otherKeyedStream)
        .between(Time.milliseconds(-2), Time.milliseconds(2)) // lower and upper bound
        .upperBoundExclusive(true) // optional
        .lowerBoundExclusive(true) // optional
        .process(new IntervalJoinFunction() { ...});

## 2.17. WindowCoGroup ##

DataStream，DataStream --> DataStream：根据一个给定的key和window对两个DataStream做CoGroups操作。

dataStream.coGroup(otherStream)
        .where(0).equalTo(1)
        .window(TumblingEventTimeWindows.of(Time.seconds(3)))
        .apply (new CoGroupFunction () { ...});

package org.apache.flink.streaming.api.test;
    
    import org.apache.flink.api.common.functions.CoGroupFunction;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.util.Collector;
    
    public class CoGroupDemo { 
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream9000 = env.socketTextStream("localhost", 9000, "\n");
            DataStream<String> textStream9001 = env.socketTextStream("localhost", 9001, "\n");
            //将输入处理一下，变为tuple2
            DataStream<Tuple2<String, String>> mapStream9000 = textStream9000
                    .map(new MapFunction<String, Tuple2<String, String>>() { 
                        @Override
                        public Tuple2<String, String> map(String s) throws Exception { 
                            return Tuple2.of(s, "来自9000端口：" + s);
                        }
                    });
    
            DataStream<Tuple2<String, String>> mapStream9001 = textStream9001
                    .map(new MapFunction<String, Tuple2<String, String>>() { 
                        @Override
                        public Tuple2<String, String> map(String s) throws Exception { 
                            return Tuple2.of(s, "来自9001端口：" + s);
                        }
                    });
    
            //3.两个流进行coGroup操作,没有关联上的也保留下来，功能更强大
            DataStream<String> result = mapStream9000.coGroup(mapStream9001)
                    //关联条件，以第0列关联（两个source输入的字符串）
                    .where(t1 -> t1.getField(0)).equalTo(t2 -> t2.getField(0))
                    //以处理时间，每10秒一个滚动窗口
                    .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                    //关联后输出
                    .apply(new CoGroupFunction<Tuple2<String, String>, Tuple2<String, String>, String>() { 
                        @Override
                        public void coGroup(Iterable<Tuple2<String, String>> iterable, Iterable<Tuple2<String, String>> iterable1, Collector<String> collector) throws Exception { 
                            StringBuffer stringBuffer = new StringBuffer();
                            stringBuffer.append("来自9000的stream:");
                            for (Tuple2<String, String> item : iterable) { 
                                stringBuffer.append(item.f1 + ",");
                            }
                            stringBuffer.append("来自9001的stream:");
                            for (Tuple2<String, String> item : iterable1) { 
                                stringBuffer.append(item.f1 + ",");
                            }
                            collector.collect(stringBuffer.toString());
                        }
                    });
    
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    }

## 2.18. Connect ##

DataStream，DataStream --> ConnectedStreams：连接两个保持她们类型的数据流。

DataStream<Integer> someStream = //...
    DataStream<String> otherStream = //...
    
    ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream);

package org.apache.flink.streaming.api.test;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.co.CoMapFunction;
    
    import java.util.regex.Matcher;
    import java.util.regex.Pattern;
    
    public class ConnectDemo { 
        public static void main(String[] args) throws Exception { 
            //1.获取执行环境配置信息
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //2.定义加载或创建数据源（source）,监听9000端口的socket消息
            DataStream<String> textStream9000 = env.socketTextStream("localhost", 9000, "\n");
            DataStream<String> textStream9001 = env.socketTextStream("localhost", 9001, "\n");
            //转为Integer类型流
            DataStream<Integer> intStream = textStream9000.filter(s -> isNumeric(s)).map(s -> Integer.valueOf(s));
            //连接起来，分别处理，返回同样的一种类型。
            SingleOutputStreamOperator result = intStream.connect(textStream9001)
                    .map(new CoMapFunction<Integer, String, Tuple2<Integer, String>>() { 
                        @Override
                        public Tuple2<Integer, String> map1(Integer value) throws Exception { 
                            return Tuple2.of(value, "");
                        }
    
                        @Override
                        public Tuple2<Integer, String> map2(String value) throws Exception { 
                            return Tuple2.of(null, value);
                        }
                    });
            //4.打印输出sink
            result.print();
            //5.开始执行
            env.execute();
        }
    
        private static boolean isNumeric(String str) { 
            Pattern pattern = Pattern.compile("[0-9]*");
            Matcher isNum = pattern.matcher(str);
            if (!isNum.matches()) { 
                return false;
            }
            return true;
        }
    }

## 2.19. CoMap 和 CoFlatMap ##

DataStream，DataStream --> DataStream：根据一个给定的key和window对两个DataStream做CoGroups操作。

connectedStreams.map(new CoMapFunction<Integer, String, Boolean>() { 
        @Override
        public Boolean map1(Integer value) { 
            return true;
        }
    
        @Override
        public Boolean map2(String value) { 
            return false;
        }
    });
    connectedStreams.flatMap(new CoFlatMapFunction<Integer, String, String>() { 
    
       @Override
       public void flatMap1(Integer value, Collector<String> out) { 
           out.collect(value.toString());
       }
    
       @Override
       public void flatMap2(String value, Collector<String> out) { 
           for (String word: value.split(" ")) { 
             out.collect(word);
           }
       }
    });

## 2.20. Split \[废弃\] ##

DataStream --> SplitStream：根据某些特征把一个DataStream拆分成两个或多个DataStream. 推荐使用sideOutput .

## 2.21. Select \[废弃\] ##

SplitStream --> DataStream：从一个SplitStream中获取一个或多个DataStream , 与Split算子配合使用.

## 2.22. Iterate ##

DataStream --> IterativeStream --> DataStream：在API层面上, 对DataStream应用个迭代会产生一个IteractiveStream , 然后再IteractiveStream应用业务处理逻辑, 最终生成一个新的DataStream, IterativeStream本质上是一种中间数据流对象.

在数据流中创建一个迭代循环,即将下游的输出发送给上游重新处理.

IterativeStream<Long> iteration = initialStream.iterate();
    DataStream<Long> iterationBody = iteration.map (/*do something*/);
    DataStream<Long> feedback = iterationBody.filter(new FilterFunction<Long>(){ 
        @Override
        public boolean filter(Long value) throws Exception { 
        	// to feedback
            return value > 0;
        }
    });
    iteration.closeWith(feedback);
    DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){ 
        @Override
        public boolean filter(Long value) throws Exception { 
        	// to next flow 
            return value <= 0;
        }
    });

## 2.23. Extract Timestamps ##

DataStream --> DataStream：提取记录中的时间, 并生成Watermark . 改算子不会改变DataStream .

## 2.24. Project ##

DataStream → DataStream： 从元组类型的数据流中抽取元组中部分元素

适用于Tuple类型的DataStream, 使用Project选取子Tuple,可以选择Tuple的部分元素,可以改变元素的顺序,类似于SQL语句中的Select子句. 输出的流仍然是DataStream

DataStream<Tuple3<Integer, Double, String>> in = // [...]
    DataStream<Tuple2<String, Integer>> out = in.project(2,0);

# 三.物理分区 #

Flink 也提供以下方法让用户根据需要在数据转换完成后对数据分区进行更细粒度的配置。

## 3.1. Custom partitioning ##

DataStream → DataStream : 使用用户定义的分区程序为每个元素选择目标任务。

dataStream.partitionCustom(partitioner, "someKey");
    dataStream.partitionCustom(partitioner, 0);

## 3.2. Random partitioning ##

DataStream → DataStream : 根据均匀分布对元素进行随机划分。

dataStream.shuffle();

## 3.3. Rebalancing (Round-robin partitioning) ##

DataStream → DataStream : 分区元素轮循，每个分区创建相等的负载。 在存在数据偏斜的情况下对性能优化有用。

dataStream.rebalance();

## 3.4. Rescaling ##

DataStream → DataStream : 将元素循环地分配到下游操作的子集。如果您希望拥有pipelines，例如从源的每个并行实例散开到几个映射器的子集以分配负载，但又不希望rebalance（）引起完全的重新平衡，则这很有用。  
这将仅需要本地数据传输，而不需要通过网络传输数据，具体取决于其他配置值，例如TaskManager的插槽数。

上游操作向其发送元素的下游操作的子集取决于上游操作和下游操作两者的并行度。  
例如，如果上游操作具有并行性2，而下游操作具有并行性6，则一个上游操作将元素分配给三个下游操作，而另一个上游操作将分配给其他三个下游操作。另一方面，如果下游操作具有并行性2，而上游操作具有并行性6，则三个上游操作将分配给一个下游操作，而其他三个上游操作将分配给另一个下游操作。

如果不同的并行度不是彼此的倍数，则一个或多个下游操作将具有与上游操作不同的输入数量。

请参见此图，以查看上面示例中的连接模式：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

dataStream.rescale();

## 3.5. Broadcasting ##

DataStream → DataStream : 将元素广播到每个分区。

dataStream.broadcast();

# 四. 算子链和资源组 #

将两个算子链接在一起能使得它们在同一个线程中执行，从而提升性能。Flink 默认会将能链接的算子尽可能地进行链接(例如， 两个 map 转换操作)。此外， Flink 还提供了对链接更细粒度控制的 API 以满足更多需求：

如果想对整个作业禁用算子链，可以调用 StreamExecutionEnvironment.disableOperatorChaining()。下列方法还提供了更细粒度的控制。需要注 意的是， 这些方法只能在 DataStream 转换操作后才能被调用，因为它们只对前一次数据转换生效。例如，可以 someStream.map(…).startNewChain() 这样调用，而不能 someStream.startNewChain()这样。

## 4.1.Start new chain ##

以当前 operator 为起点开始新的连接。如下的两个 mapper 算子会链接在一起而 filter 算子则不会和第一个 mapper 算子进行链接。

> someStream.filter(…).map(…).startNewChain().map(…);

## 4.2.Disable chaining ##

任何算子不能和当前算子进行链接

> someStream.map(…).disableChaining();

## 4.3.Set slot sharing group ##

配置算子的资源组。  
**Flink 会将相同资源组的算子放置到同一个 slot 槽中执行，并将不同资源组的算子分配到不同的 slot 槽中，从而实现 slot 槽隔离。**  
**资源组将从输入算子开始继承如果所有输入操作都在同一个资源组。**  
**Flink 默认的资源组名称为 “default”，算子可以显式调用 slotSharingGroup(“default”) 加入到这个资源组中。**

> someStream.filter(…).slotSharingGroup(“name”);

参考文章:  
https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/stream/operators/  
https://www.jianshu.com/p/4183dfaa7904  
https://blog.csdn.net/chybin500/article/details/87260869  
https://ververica.cn/developers/apache-flink-basic-zero-iii-datastream-api-programming/  
Flink内核原理与实现 -冯飞

[Flink]: /images/20221024/03c8388480b24691b071ce3b1727cf4e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221024/4b546384c21f4f44bc80d1dacfe5b142.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221024/a6a1350c0e9d4db485eab86752ca9eea.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20221024/5fe208c914344ca9abe87bb01ef31cf5.png