Flink DataStream API 基础算子（一）

一、介绍

官网

DataStream API 得名于特殊的 DataStream 类，该类用于表示 Flink 程序中的数据集合。你可以认为它们是可以包含重复项的不可变数据集合。这些数据可以是有界（有限）的，也可以是无界（无限）的，但用于处理它们的API是相同的。

二、基础算子

1、Map

Map算子：输入一个元素同时输出一个元素,这里的写法和Java类似,可以使用糖化语法或者实现Function接口

package com.xx.common.study.api.base;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** @author xiaxing* @describe Map算子,输入一个元素同时输出一个元素,这里的写法和Java类似,可以使用糖化语法或者实现Function接口* @since 2024/5/17 14:27*/
public class DataStreamMapApiDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<Integer> sourceStream = env.fromElements(1, 2, 3);// 糖化语法SingleOutputStreamOperator<Integer> multiStream = sourceStream.map(e -> e * 2);multiStream.print("数据乘2");// 实现Function接口SingleOutputStreamOperator<Integer> addStream = sourceStream.map(new MapFunction<Integer, Integer>() {@Overridepublic Integer map(Integer value) throws Exception {return value + 2;}});addStream.print("数据加2");env.execute();}
}

2、FlatMap

FlatMap算子：输入一个元素同时产生零个、一个或多个元素

package com.xx.common.study.api.base;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Arrays;
import java.util.List;/*** @author xiaxing* @describe FlatMap算子,输入一个元素同时产生零个、一个或多个元素* @since 2024/5/17 14:27*/
public class DataStreamFlatMapApiDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> sourceStream = env.fromElements("1,2,3");// 对source进行加工处理sourceStream.flatMap((FlatMapFunction<String, List<String>>) (value, out) -> {String[] split = value.split(",");out.collect(Arrays.asList(split));}).print();// 错误写法,和Java写法不用,无法使用这种糖化语法
//        sourceStream.flatMap((k, v) -> {
//            String[] split = k.split(",");
//            v.collect(split);
//        }).print();env.execute();}
}

3、Filter

Filter算子：为每个元素执行一个布尔 function，并保留那些 function 输出值为 true 的元素

package com.xx.common.study.api.base;import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** @author xiaxing* @describe Filter算子,为每个元素执行一个布尔 function，并保留那些 function 输出值为 true 的元素* @since 2024/5/17 14:27*/
public class DataStreamFilterApiDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<Integer> sourceStream = env.fromElements(1, 2, 3);// 保留整数sourceStream.filter(e -> (e % 2) == 0).print("糖化语法保留整数");sourceStream.filter(new FilterFunction<Integer>() {@Overridepublic boolean filter(Integer value) throws Exception {return value % 2 == 0;}}).print("实现Function保留整数");env.execute();}
}

4、KeyBy

KeyBy算子：在逻辑上将流划分为不相交的分区。具有相同 key 的记录都分配到同一个分区。在内部， keyBy() 是通过哈希分区实现的

package com.xx.common.study.api.base;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** @author xiaxing* @describe KeyBy算子,在逻辑上将流划分为不相交的分区。具有相同 key 的记录都分配到同一个分区。在内部， keyBy() 是通过哈希分区实现的* @since 2024/5/17 14:27*/
public class DataStreamKeyByApiDemo {@Data@AllArgsConstructor@NoArgsConstructorpublic static class keyByDemo {private Integer id;private Integer count;}public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<keyByDemo> sourceStream = env.fromElements(new keyByDemo(1, 1),new keyByDemo(2, 2),new keyByDemo(3, 3),new keyByDemo(1, 4));KeyedStream<keyByDemo, Integer> keyByStream = sourceStream.keyBy(keyByDemo::getId);keyByStream.print("按照key分组");// 使用key分组之后可以使用一些常用的聚合算子// positionToSum:可以用于Tuple类型数据传递索引位置,field:传递字段名称keyByStream.sum("count").print();env.execute();}
}

5、Reduce

Reduce算子：在相同 key 的数据流上“滚动”执行 reduce。将当前元素与最后一次 reduce 得到的值组合然后输出新值

package com.xx.common.study.api.base;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** @author xiaxing* @describe Reduce算子,在相同 key 的数据流上“滚动”执行 reduce。将当前元素与最后一次 reduce 得到的值组合然后输出新值* @since 2024/5/17 14:27*/
public class DataStreamReduceApiDemo {@Data@AllArgsConstructor@NoArgsConstructorpublic static class reduceByDemo {private String id;private Integer count;}public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<reduceByDemo> sourceStream = env.fromElements(new reduceByDemo("1", 1),new reduceByDemo("2", 2),new reduceByDemo("3", 3),new reduceByDemo("1", 4));sourceStream.keyBy(reduceByDemo::getId).reduce(new ReduceFunction<reduceByDemo>() {@Overridepublic reduceByDemo reduce(reduceByDemo value1, reduceByDemo value2) throws Exception {value1.setCount(value1.getCount() + value2.getCount());return value1;}}).print();env.execute();}
}

三、窗口算子

官网地址

3.1、概念

窗口（Window）是处理无界流的关键所在。窗口可以将数据流装入大小有限的“桶”中，再对每个“桶”加以处理。

Flink 中的时间有三种类型：

Event Time：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink 通过时间戳分配器访问事件时间戳。
Ingestion Time：是数据进入 Flink 的时间。
Processing Time：是每一个执行基于时间操作的算子的本地系统时间，与机器相关，默认的时间属性就是 Processing Time。

3.2、语法

Keyed Windows

stream.keyBy(...)               <-  仅 keyed 窗口需要.window(...)              <-  必填项："assigner"[.trigger(...)]            <-  可选项："trigger" (省略则使用默认 trigger)[.evictor(...)]            <-  可选项："evictor" (省略则不使用 evictor)[.allowedLateness(...)]    <-  可选项："lateness" (省略则为 0)[.sideOutputLateData(...)] <-  可选项："output tag" (省略则不对迟到数据使用 side output).reduce/aggregate/apply()      <-  必填项："function"[.getSideOutput(...)]      <-  可选项："output tag"

Non-Keyed Windows

stream.windowAll(...)           <-  必填项："assigner"[.trigger(...)]            <-  可选项："trigger" (else default trigger)[.evictor(...)]            <-  可选项："evictor" (else no evictor)[.allowedLateness(...)]    <-  可选项："lateness" (else zero)[.sideOutputLateData(...)] <-  可选项："output tag" (else no side output for late data).reduce/aggregate/apply()      <-  必填项："function"[.getSideOutput(...)]      <-  可选项："output tag"

3.3、Window Assigners

Window Assigners为抽象类，Flink默认已经实现了4种窗口

3.3.1、滚动窗口（Tumbling Windows）

滚动窗口的 assigner 分发元素到指定大小的窗口。滚动窗口的大小是固定的，且各自范围之间不重叠。比如说，如果你指定了滚动窗口的大小为 5 分钟，那么每 5 分钟就会有一个窗口被计算，且一个新的窗口被创建（如下图所示）。

在这里插入图片描述

DataStream<T> input = ...;// 滚动 event-time 窗口
input.keyBy(<key selector>).window(TumblingEventTimeWindows.of(Time.seconds(5))).<windowed transformation>(<window function>);// 滚动 processing-time 窗口
input.keyBy(<key selector>).window(TumblingProcessingTimeWindows.of(Time.seconds(5))).<windowed transformation>(<window function>);// 长度为一天的滚动 event-time 窗口， 偏移量为 -8 小时。
input.keyBy(<key selector>).window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8))).<windowed transformation>(<window function>);

3.3.2、滑动窗口（Sliding Windows）

与滚动窗口类似，滑动窗口的 assigner 分发元素到指定大小的窗口，窗口大小通过 window size 参数设置。滑动窗口需要一个额外的滑动距离（window slide）参数来控制生成新窗口的频率。因此，如果 slide 小于窗口大小，滑动窗口可以允许窗口重叠。这种情况下，一个元素可能会被分发到多个窗口。

比如说，你设置了大小为 10 分钟，滑动距离 5 分钟的窗口，你会在每 5 分钟得到一个新的窗口，里面包含之前 10 分钟到达的数据（如下图所示）。

在这里插入图片描述

DataStream<T> input = ...;// 滑动 event-time 窗口
input.keyBy(<key selector>).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))).<windowed transformation>(<window function>);// 滑动 processing-time 窗口
input.keyBy(<key selector>).window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5))).<windowed transformation>(<window function>);// 滑动 processing-time 窗口，偏移量为 -8 小时
input.keyBy(<key selector>).window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8))).<windowed transformation>(<window function>);

3.3.3、会话窗口（Session Windows）

会话窗口的 assigner 会把数据按活跃的会话分组。与滚动窗口和滑动窗口不同，会话窗口不会相互重叠，且没有固定的开始或结束时间。会话窗口在一段时间没有收到数据之后会关闭，即在一段不活跃的间隔之后。会话窗口的 assigner 可以设置固定的会话间隔（session gap）或用 session gap extractor 函数来动态地定义多长时间算作不活跃。当超出了不活跃的时间段，当前的会话就会关闭，并且将接下来的数据分发到新的会话窗口。

在这里插入图片描述

DataStream<T> input = ...;// 设置了固定间隔的 event-time 会话窗口
input.keyBy(<key selector>).window(EventTimeSessionWindows.withGap(Time.minutes(10))).<windowed transformation>(<window function>);// 设置了动态间隔的 event-time 会话窗口
input.keyBy(<key selector>).window(EventTimeSessionWindows.withDynamicGap((element) -> {// 决定并返回会话间隔})).<windowed transformation>(<window function>);// 设置了固定间隔的 processing-time session 窗口
input.keyBy(<key selector>).window(ProcessingTimeSessionWindows.withGap(Time.minutes(10))).<windowed transformation>(<window function>);// 设置了动态间隔的 processing-time 会话窗口
input.keyBy(<key selector>).window(ProcessingTimeSessionWindows.withDynamicGap((element) -> {// 决定并返回会话间隔})).<windowed transformation>(<window function>);

3.3.4、全局窗口（Global Windows）

全局窗口的 assigner 将拥有相同 key 的所有数据分发到一个全局窗口。这样的窗口模式仅在你指定了自定义的 trigger 时有用。否则，计算不会发生，因为全局窗口没有天然的终点去触发其中积累的数据。

在这里插入图片描述

DataStream<T> input = ...;input.keyBy(<key selector>).window(GlobalWindows.create()).<windowed transformation>(<window function>);

3.4、窗口函数

窗口函数有三种：ReduceFunction、AggregateFunction 或 ProcessWindowFunction

3.4.1、ReduceFunction

ReduceFunction 指定两条输入数据如何合并起来产生一条输出数据，输入和输出数据的类型必须相同

package com.xx.common.study.api.windows;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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;/*** @author xiaxing* @describe 窗口函数-reduce* @since 2024/5/17 14:27*/
public class DataStreamWindowsReduceApiDemo {@Data@NoArgsConstructor@AllArgsConstructorpublic static class TumblingWindows {private Integer id;private Integer count;}public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> socketStream = env.socketTextStream("localhost", 7777);socketStream.map(new MapFunction<String, TumblingWindows>() {@Overridepublic TumblingWindows map(String value) throws Exception {String[] split = value.split(",");return new TumblingWindows(Integer.valueOf(split[0]), Integer.valueOf(split[1]));}}).keyBy(new KeySelector<TumblingWindows, Integer>() {@Overridepublic Integer getKey(TumblingWindows value) throws Exception {return value.getId();}}).window(TumblingProcessingTimeWindows.of(Time.seconds(5L))).reduce(new ReduceFunction<TumblingWindows>() {@Overridepublic TumblingWindows reduce(TumblingWindows value1, TumblingWindows value2) throws Exception {return new TumblingWindows(value1.getId(), value1.getCount() + value2.getCount());}}).print();env.execute();}
}

3.4.2、ReduceFunction糖化语法

使用Lambda糖化语法对代码进行了简化

package com.xx.common.study.api.windows;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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;/*** @author xiaxing* @describe 窗口函数-reduce-糖化语法* @since 2024/5/17 14:27*/
public class DataStreamWindowsReduceLambdaApiDemo {@Data@NoArgsConstructor@AllArgsConstructorpublic static class TumblingWindows {private Integer id;private Integer count;}public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> socketStream = env.socketTextStream("localhost", 7777);socketStream.map(value -> {String[] split = value.split(",");return new TumblingWindows(Integer.valueOf(split[0]), Integer.valueOf(split[1]));}).keyBy(TumblingWindows::getId).window(TumblingProcessingTimeWindows.of(Time.seconds(5L))).reduce((value1, value2) -> new TumblingWindows(value1.getId(), value1.getCount() + value2.getCount())).print();env.execute();}
}

3.4.3、AggregateFunction

ReduceFunction 是 AggregateFunction 的特殊情况。 AggregateFunction 接收三个类型：输入数据的类型(IN)、累加器的类型（ACC）和输出数据的类型（OUT）。输入数据的类型是输入流的元素类型，AggregateFunction 接口有如下几个方法：把每一条元素加进累加器、创建初始累加器、合并两个累加器、从累加器中提取输出（OUT 类型）

package com.xx.common.study.api.windows;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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 java.util.Optional;/*** @author xiaxing* @describe 窗口函数-Aggregate* @since 2024/5/17 14:27*/
public class DataStreamWindowsAggregateApiDemo {@Data@NoArgsConstructor@AllArgsConstructorpublic static class TumblingWindows {private Integer id;private Integer count;}public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> socketStream = env.socketTextStream("localhost", 7777);// 求和AggregateFunction<TumblingWindows, TumblingWindows, TumblingWindows> aggregateFunction = new AggregateFunction<TumblingWindows, TumblingWindows, TumblingWindows>() {@Overridepublic TumblingWindows createAccumulator() {// 创建累加器,并将其初始化为默认值return new TumblingWindows();}@Overridepublic TumblingWindows add(TumblingWindows value, TumblingWindows accumulator) {// 将输入的元素添加到累加器,返回更新后的累加器Integer count1 = Optional.of(value.getCount()).orElse(0);Integer count2 = Optional.ofNullable(accumulator.getCount()).orElse(0);return new TumblingWindows(value.getId(), count1 + count2);}@Overridepublic TumblingWindows getResult(TumblingWindows accumulator) {// 从累加器中提取操作的结果return accumulator;}@Overridepublic TumblingWindows merge(TumblingWindows a, TumblingWindows b) {// 将两个累加器合并为一个新的累加器return new TumblingWindows(a.getId(), a.getCount() + b.getCount());}};socketStream.map(value -> {String[] split = value.split(",");return new TumblingWindows(Integer.valueOf(split[0]), Integer.valueOf(split[1]));}).keyBy(TumblingWindows::getId).window(TumblingProcessingTimeWindows.of(Time.seconds(5L))).aggregate(aggregateFunction).print();env.execute();}
}

3.4.4、ProcessWindowFunction

ProcessWindowFunction 可以与 ReduceFunction 或 AggregateFunction 搭配使用，使其能够在数据到达窗口的时候进行增量聚合。当窗口关闭时，ProcessWindowFunction 将会得到聚合的结果。这样它就可以增量聚合窗口的元素并且从 ProcessWindowFunction` 中获得窗口的元数据。

package com.xx.common.study.api.windows;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;/*** @author xiaxing* @describe 窗口函数-reduce-process* @since 2024/5/17 14:27*/
public class DataStreamWindowsReduceProcessApiDemo {@Data@NoArgsConstructor@AllArgsConstructorpublic static class TumblingWindows {private Integer id;private Integer count;}public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> socketStream = env.socketTextStream("localhost", 7777);socketStream.map(value -> {String[] split = value.split(",");return new TumblingWindows(Integer.valueOf(split[0]), Integer.valueOf(split[1]));}).keyBy(TumblingWindows::getId).window(TumblingProcessingTimeWindows.of(Time.seconds(5L))).reduce(new MyReduceFunction(), new MyProcessWindowsFunction()).print();env.execute();}private static class MyReduceFunction implements ReduceFunction<TumblingWindows> {@Overridepublic TumblingWindows reduce(TumblingWindows value1, TumblingWindows value2) throws Exception {return new TumblingWindows(value1.getId(), value1.getCount() + value2.getCount());}}private static class MyProcessWindowsFunction extends ProcessWindowFunction<TumblingWindows, TumblingWindows, Integer, TimeWindow> {@Overridepublic void process(Integer integer, ProcessWindowFunction<TumblingWindows, TumblingWindows, Integer, TimeWindow>.Context context, Iterable<TumblingWindows> elements, Collector<TumblingWindows> out) throws Exception {elements.forEach(e -> {Integer count = e.getCount();// 当count > 10时才数据元素if (count > 10) {out.collect(e);}});}}
}