[大数据]java Flink（三十九）官方基础篇-算子概括（基础）

? 今天回顾一下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()是通过哈希分区实现的。有多种指定键的方法。

? 使用方式：

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