java Flink（三十九）官方基础篇-算子概括（基础）

男娘i 2022-08-31 06:58 9阅读 0赞

**  今天回顾一下Flink官方对内部提供的算子的一个介绍。**

**算子：Map**

**  转换流：**DataStream → DataStream

**  简单说明：**取一个元素并产生一个元素。将输入流的值加倍的映射函数（必须产生数据）

**  使用方式：**

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

**算子：FlatMap**

**  转换流：**DataStream → DataStream

**  简单说明：**取一个元素并产生零个、一个或多个元素。将句子拆分为单词的 flatmap 函数（不一定会产生数据）

**  使用方式：**

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);
            }
        }
    });

**算子：Filter**

**  转换流：**DataStream → DataStream

**  简单说明：**为每个元素计算一个布尔函数，并保留那些函数返回 true 的元素。过滤掉零值的过滤器

**  使用方式：**

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

**算子：KeyBy**

**  转换流：**DataStream → KeyedStream

**  简单说明：**在逻辑上将流划分为不相交的分区。所有具有相同键的记录都分配到同一个分区。在内部，*keyBy()*是通过哈希分区实现的。有多种[指定键的方法][Link 1]。

**  使用方式：**

dataStream.keyBy(value -> value.getSomeKey());
    dataStream.keyBy(value -> value.f0);

**算子：Reduce**

**  转换流：**KeyedStream → DataStream

**  简单说明：**键控数据流上的“滚动”减少。将当前元素与最后的值组合并发出新值。

**  使用方式：**

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

**算子：Window **

**  转换流：**KeyedStream → WindowedStream

**  简单说明：**可以在已经分区的 KeyedStreams 上定义 Windows。Windows 根据某些特征（例如，最近 5 秒内到达的数据）对每个键中的数据进行分组

**  使用方式：**

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

**算子：WindowAll **

**  转换流：**DataStreamStream → AllWindowedStream

**  简单说明：**可以在常规数据流上定义 Windows。Windows 根据某些特征（例如，在过去 5 秒内到达的数据）对所有流事件进行分组。

**  使用方式：**

dataStream
      .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));

**算子：Window Apply**

**  转换流：**

WindowedStream → DataStream （可以用于keyby后）

AllWindowedStream → DataStream

**  简单说明：**将通用函数应用于整个窗口。

**  使用方式：**

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));
        }
    });

**算子：WindowReduce **

**  转换流：**WindowedStream → DataStream

**  简单说明：**对窗口应用函数式缩减函数并返回缩减后的值。

**  使用方式：**

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);
        }
    });

**算子：Union **

**  转换流：**DataStream\* → DataStream

**  简单说明：**两个或多个数据流的联合创建一个包含所有流中所有元素的新流。注意：如果您将数据流与自身联合，您将在结果流中获得每个元素两次。

**  使用方式：**

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

**算子：Window Join**

**  转换流：**DataStream,DataStream → DataStream

**  简单说明：**在给定的键和公共窗口上连接两个数据流。

**  使用方式：**

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

**算子：Interval Join**

**  转换流：**KeyedStream,KeyedStream → DataStream

**  简单说明：**在给定的时间间隔内使用公共密钥连接两个带键流的两个元素 e1 和 e2，使得`e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound`.

**  使用方式：**

// 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() {...});

**算子：Window CoGroup**

**  转换流：**DataStream,DataStream → DataStream

**  简单说明：**将给定键和公共窗口上的两个数据流组合在一起。

**  使用方式：**

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

[Link 1]: https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/datastream/fault-tolerance/state/#keyed-datastream