之前的流处理API,无论是基本的转换、聚合,还是更为复杂的窗口操作,其实都是基于DataStream进行转换的,所以可以统称为DataStream API。
在Flink更底层,我们可以不定义任何具体的算子(比如map,filter,或者window),而只是提炼出一个统一的“处理”(process)操作——它是所有转换算子的一个概括性的表达,可以自定义处理逻辑,所以这一层接口就被叫作“处理函数”(process function)。
一.基本处理函数(ProcessFunction)
1.1 处理函数的功能和使用
转换算子一般针对某种具体操作来定义的,能拿到的信息有限。而使用底层的处理函数,则可以使用处理函数提供的“定时服务”(TimerServer) 来获取到当前当前水位线、事件等更为详细的信息,及注册“定时事件”。而且处理函数继承了AbstractRichFunction抽象类,所以拥有富函数类的所有特性,同样可以访问状态(state)和其他运行时信息。此外,处理函数还可以直接将数据输出到侧输出流(side output)中。所以,处理函数是最为灵活的处理方法,可以实现各种自定义的业务逻辑。
1.2 ProcessFunction解析
ProcessFunction的使用就基于 DataStream 调用 .process() 方法传入一个ProcessFunction作为参数,用来定义处理逻辑。
从源码可以看到,ProcessFunction 继承了 AbstractRichFunction(抽象富函数类),两个泛类型参数代表输入类型与输出类型。里面单独定于了两个非常重要的方法:一个是必须要实现的抽象方法.processElement();另一个是非抽象方法.onTimer()。
- processElement():具体的数据处理逻辑,且对于流中的每个元素都会调用一次。三个参数分别为 value当前数据、ctx上下文、out采集器。
- onTimer():当注册的定时器被触发,会执行该方法。三个参数分别为 timestamp时间戳、ctx上下文、out采集器。在Flink中,只有“按键分区流”KeyedStream才支持设置定时器的操作。
通过几个参数的分析不难发现,ProcessFunction可以轻松实现flatMap、map、filter这样的基本转换功能;而通过富函数提供的获取上下文方法.getRuntimeContext(),也可以自定义状态(state)进行处理,这也就能实现聚合操作的功能了。
1.3 处理函数的分类
DataStream在调用一些转换方法之后,有可能生成新的流类型;例如调用 .KeyBy() 后得到的是KeyedStream,然后调用 .window() 后得到的 WindowedStream。但是对于不同的流类型,都可以直接调用 .process() 方法进行自定义处理,此时传入的参数就叫做处理函数。当然,它们尽管本质相同,都是可以访问状态和时间信息的底层API,可彼此之间也会有所差异。
Flink提供了8个不同的处理函数:
(1) ProcessFunction
最基本的处理函数,基于DataStream直接调用.process()时作为参数传入。
(2) KeyedProcessFunction
对流按键分区后的处理函数,基于KeyedStream调用.process()时作为参数传入。要想使用定时器,比如基于KeyedStream。
(3) ProcessWindowFunction
开窗之后的处理函数,也是全窗口函数的代表。基于WindowedStream调用.process()时作为参数传入。
(4) ProcessAllWindowFunction
同样是开窗之后的处理函数,基于AllWindowedStream调用.process()时作为参数传入。
(5) CoProcessFunction
合并(connect)两条流之后的处理函数,基于ConnectedStreams调用.process()时作为参数传入。
(6) ProcessJoinFunction
间隔连接(interval join)两条流之后的处理函数,基于IntervalJoined调用.process()时作为参数传入。
(7) BroadcastProcessFunction
广播连接流处理函数,基于BroadcastConnectedStream调用.process()时作为参数传入。这里的“广播连接流”BroadcastConnectedStream,是一个未keyBy的普通DataStream与一个广播流(BroadcastStream)做连接(conncet)之后的产物。
(8) KeyedBroadcastProcessFunction
按键分区的广播连接流处理函数,同样是基于BroadcastConnectedStream调用.process()时作为参数传入。与BroadcastProcessFunction不同的是,这时的广播连接流,是一个KeyedStream与广播流(BroadcastStream)做连接之后的产物。
二.按键分区处理函数(KeyedProcessFunction)
上面提到,只有在KeyedStream中才支持使用TimerService设置定时器的操作。所以我们一般是先对流做 KeyBy 分区操作后,再去调用 .process() 定义具体的操作逻辑逻辑;一般传入 KeyedProcessFunction。
2.1 定时器(Timer)和定时服务(TimerService)
在.onTimer()方法中可以实现定时处理的逻辑,当之前注册的定时器被触发,则会调用该方法。注册定时器的功能,是通过上下文中提供的“定时服务”来实现的。
通过KeyedProcessFunction提供的上下文可以获取以下等内容:
ds
.keyBy( t -> t.getId())
.process(new KeyedProcessFunction<String, WaterSensor, Object>() {
/**
* 来一条数据调用一次
* @param value 当前输入的数据
* @param ctx 上下文信息
* @param out 采集器
* @throws Exception
*/
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, Object>.Context ctx, Collector<Object> out) throws Exception {
// 获取定时服务
TimerService timerService = ctx.timerService();
// 注册定时器:以事件时间为基准
timerService.registerEventTimeTimer(long time);
// 注册定时器:以处理时间为基准
timerService.registerProcessingTimeTimer(long time);
// 当前的处理时间:即系统时间
timerService.currentProcessingTime();
// 删除触发时间为time的处事件时间定时器
timerService.deleteEventTimeTimer(long time);
// 删除触发时间为time的处理时间定时器
timerService.deleteEventTimeTimer(long time);
// 获取当前水位线 ***获取的上一条数据的水位线
timerService.currentWatermark();
}
/**
* 时间进展到定时器注册的时间则会调用该方法
* @param timestamp 当前时间戳
* @param ctx 上下文信息
* @param out 采集器
* @throws Exception
*/
@Override
public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, Object>.OnTimerContext ctx, Collector<Object> out) throws Exception {
}
});
TimerService会以键(key)和时间戳为标准,对定时器进行去重;也就是说对于每个key和时间戳,最多只有一个定时器,如果注册了多次,onTimer()方法也将只被调用一次。
2.2 KeyedProcessFunction案例
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl());
// ***定义 WaterMark 策略
WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L);
// ***指定 watermark策略
SingleOutputStreamOperator<WaterSensor> sensorWithWaterMark = sensorDS
.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy);
sensorWithWaterMark
.keyBy( t -> t.getId())
.process(new KeyedProcessFunction<String, WaterSensor, Object>() {
/**
* 来一条数据调用一次
* @param value 当前输入的数据
* @param ctx 上下文信息
* @param out 采集器
* @throws Exception
*/
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, Object>.Context ctx, Collector<Object> out) throws Exception {
// 获取定时服务
TimerService timerService = ctx.timerService();
// 当前 Key
String currentKey = ctx.getCurrentKey();
// 数据中的事件时间
Long timestamp = ctx.timestamp();
// 注册以事件时间为基准的5s定时器
timerService.registerEventTimeTimer(5000);
System.out.println("当前 Key 为=" + currentKey + "当前时间为=" + timestamp + "注册了一个5s的定时器");
}
/**
* 时间进展到定时器注册的时间则会调用该方法
* @param timestamp 当前时间戳
* @param ctx 上下文信息
* @param out 采集器
* @throws Exception
*/
@Override
public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, Object>.OnTimerContext ctx, Collector<Object> out) throws Exception {
// 当前 Key
String currentKey = ctx.getCurrentKey();
System.out.println("当前 Key 为=" + currentKey + "现在时间为" + timestamp + "定时器触发");
}
});
env.execute();
}
输入:
[root@VM-55-24-centos ~]# nc -lk 1234
S1,1,1
S1,4,4
S2,3,3
S2,5,5
S3,8,8
S3,9,9
输出:
当前 Key 为=S1当前时间为=1000注册了一个5s的定时器
当前 Key 为=S1当前时间为=4000注册了一个5s的定时器
当前 Key 为=S2当前时间为=3000注册了一个5s的定时器
当前 Key 为=S2当前时间为=5000注册了一个5s的定时器
当前 Key 为=S3当前时间为=8000注册了一个5s的定时器
当前 Key 为=S3当前时间为=9000注册了一个5s的定时器 // 触发定时器 9000ms-3000ms-1ms = 5999
当前 Key 为=S1现在时间为5000定时器触发
当前 Key 为=S3现在时间为5000定时器触发
当前 Key 为=S2现在时间为5000定时器触发
2.3 KeyedProcessFunction中的当前Watermark
在实现的 processElement() 中获取当前水位线
sensorWithWaterMark
.keyBy( t -> t.getId())
.process(new KeyedProcessFunction<String, WaterSensor, Object>() {
/**
* 来一条数据调用一次
* @param value 当前输入的数据
* @param ctx 上下文信息
* @param out 采集器
* @throws Exception
*/
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, Object>.Context ctx, Collector<Object> out) throws Exception {
// 获取定时服务
TimerService timerServiceService = ctx.timerService();
// 获取当前水位线
long currentWatermark = timerServiceService.currentWatermark();
System.out.println("当前数据为=" + value + "当前水位线为=" + currentWatermark );
}
});
输入:
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
s1,3,3
s1,6,6
s1,8,8
输出:
当前数据为=WaterSensor{id='s1', ts=1, vc=1}当前水位线为=-9223372036854775808
当前数据为=WaterSensor{id='s1', ts=3, vc=3}当前水位线为=-2001
当前数据为=WaterSensor{id='s1', ts=6, vc=6}当前水位线为=-1
当前数据为=WaterSensor{id='s1', ts=8, vc=8}当前水位线为=2999
可以看到,在process中的当前水位线其实是 上一条数据的事件时间 - 水位线延迟时间 - 1ms。
2.4 KeyedProcessFunction 小结
- 定时服务(TimerServer)只有 KeyedStream(键控流) 才能使用
- 事件时间定时器,是通过 Watermark 触发的
- Watermark = 当前最大事件时间 - Watermark 延迟时间 - 1ms
- 在 process 中获取到的 Watermark 其实是上一条数据的 Watermark
其他处理函数类似。
三. 应用案例——Top N
案例需求:实时统计一段时间内的出现次数最多的水位。例如,统计最近10秒钟内出现次数最多的两个水位,并且每5秒钟更新一次。我们知道,这可以用一个滑动窗口来实现。于是就需要开滑动窗口收集传感器的数据,按照不同的水位进行统计,而后汇总排序并最终输出前两名。这其实就是著名的“Top N”问题。
3.1 使用ProcessAllWindowFunction
思路:直接开窗,使用全窗口函数处理窗口内所有的数据,使用HashMap存储,再对map进行统计排序输出。统计十秒内数据,五秒输出一次,其实就是滑动窗口大小为10,滑动步长为5。
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl());
WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L);
SingleOutputStreamOperator<WaterSensor> sensorWithWaterMark = sensorDS
.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy);
sensorWithWaterMark
// 滑动窗口,窗口大小10s,滑动步长5s
.windowAll(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5)))
.process(new MyProcessAllWindowFunction())
.print();
env.execute();
}
/**
* 自定义全窗口处理函数
* 全窗口函数:窗口触发时调用一次
*/
public static class MyProcessAllWindowFunction extends ProcessAllWindowFunction<WaterSensor , String , TimeWindow>{
@Override
public void process(ProcessAllWindowFunction<WaterSensor, String, TimeWindow>.Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
// 定义一个hashmap用来存,key=vc,value=count值
Map<Integer, Integer> vcCountMap = new HashMap<>();
for (WaterSensor element : elements) {
Integer vc = element.getVc();
if (vcCountMap.containsKey(vc)) {
vcCountMap.put(vc, vcCountMap.get(vc) + 1);
} else {
vcCountMap.put(vc, 1);
}
}
List<Map.Entry<Integer, Integer>> list = new ArrayList<>(vcCountMap.entrySet());
// 使用Collections.sort()按value降序排序
Collections.sort(list, (o1, o2) -> o2.getValue() - o1.getValue());
System.out.println(list);
StringBuffer result = new StringBuffer();
for (int i = 0; i < Math.min(2 , list.size()); i++) {
result.append("Top " + (i+1) + ": vc = " + list.get(i).getKey() + ",出现次数 = " + list.get(i).getValue());
result.append("\n");
}
out.collect(result.toString());
}
}
}
输入:
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
s2,3,1
s1,5,2
s3,6,2
s1,6,1
s2,7,3
s3,8,1
s1,8,3
s2,9,2
s1,10,1
s3,11,2
s1,13,2
输出:
// 注释:[ 0 , 5 ] 的窗口
Top 1: vc = 1,出现次数 = 2
// 注释:[ 0 , 10 ] 的窗口
Top 1: vc = 1,出现次数 = 4
Top 2: vc = 2,出现次数 = 3
3.2 使用KeyedProcessFunction
上面的方法使用全窗口将所有的数据都放在一个分区内,强行将并行度设置成了1,这是Flink不推荐的做法。
则可以使用KeyedProcessFunction进行优化:
1.对统计字段(vc)进行 KeyBy 分区
2.进行增量聚合,统计vc出现的次数,封装数据(vc,count,窗口标记(窗口结束数据))
3.对标记(窗口)进行分组,对数据进行排序、取TopN
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl())
.assignTimestampsAndWatermarks(WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L)
);
// 对 vc 进行分组,统计窗口内vc出现的次数 将每条数据封装成 vc,count,窗口结束时间
SingleOutputStreamOperator<Tuple3<Integer, Integer, Long>> windowAgg = sensorDS
.keyBy(sensor -> sensor.getVc())
.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
.aggregate(
new VcCountAgg(),
new WindowResult()
);
// 按照打的标记(windowEndTime)进行分组,保证同一个窗口时间范围的结果一起 , 分组后排序,取 TopN
windowAgg.keyBy(v -> v.f2)
.process(new TopN(2))
.print();
env.execute();
}
/**
* 增量聚合:累计同分组出现的次数
*/
public static class VcCountAgg implements AggregateFunction<WaterSensor,Integer,Integer> {
@Override
public Integer createAccumulator() {
return 0;
}
@Override
public Integer add(WaterSensor waterSensor, Integer integer) {
return integer+1;
}
@Override
public Integer getResult(Integer integer) {
return integer;
}
@Override
public Integer merge(Integer integer, Integer acc1) {
return null;
}
}
/**
* 全窗口函数,窗口内数据全部到达才会执行一次
* 泛型如下:
* 第一个:输入类型 = 增量函数的输出 count值,Integer
* 第二个:输出类型 = Tuple3(vc,count,windowEndTime) ,带上 窗口结束时间 的标签
* 第三个:key类型 , vc,Integer
* 第四个:窗口类型
*/
public static class WindowResult extends ProcessWindowFunction<Integer, Tuple3<Integer, Integer, Long>, Integer, TimeWindow> {
@Override
public void process(Integer vc, ProcessWindowFunction<Integer, Tuple3<Integer, Integer, Long>, Integer, TimeWindow>.Context context, Iterable<Integer> elements, Collector<Tuple3<Integer, Integer, Long>> out) throws Exception {
// 获取迭代器的数据(只有一条数据)
Integer count = elements.iterator().next();
// 获取窗口结束时间 作为标记
long endTime = context.window().getEnd();
// 将 vc、vc对应的count 连带窗口标记 返回
out.collect(Tuple3.of(vc , count , endTime));
}
}
/**
* 处理组内的每一条数据 一条数据触发一次
*/
public static class TopN extends KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String> {
// 存不同窗口的 统计结果,key=windowEnd,value=list数据
private Map<Long, List<Tuple3<Integer, Integer, Long>>> dataListMap;
// 要取的Top数量
private int threshold;
public TopN(int threshold) {
this.threshold = threshold;
dataListMap = new HashMap<>();
}
@Override
public void processElement(Tuple3<Integer, Integer, Long> value, KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.Context ctx, Collector<String> out) throws Exception {
// 进入这个方法,只是一条数据,要排序,得到齐才行 ===》 存起来,不同窗口分开存
Long windowEnd = value.f2;
// 将对应的窗口的数据放入map中对应key的list中
if(dataListMap.containsKey(windowEnd)){
// 该 vc 存在,则直接添加到数组中
dataListMap.get(windowEnd).add(value);
}else{
// 不包含vc,是该vc的第一条
List<Tuple3<Integer, Integer, Long>> dataList = new ArrayList<>();
dataList.add(value);
dataListMap.put(windowEnd , dataList);
}
// 注册一个定时器 窗口触发时触发定时器(windowEndTime + 1ms) 输出计算结果
// 同一个窗口范围,应该同时输出,只不过是一条一条调用processElement方法,只需要延迟1ms即可
ctx.timerService().registerProcessingTimeTimer(windowEnd + 1);
}
// 注册一个定时器 窗口触发时进行排序,取TopN,输出结果
@Override
public void onTimer(long timestamp, KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
super.onTimer(timestamp, ctx, out);
// 定时器触发,同一个窗口范围的计算结果攒齐了,开始 排序、取TopN
Long windowEnd = ctx.getCurrentKey();
// 1. 排序
List<Tuple3<Integer, Integer, Long>> dataList = dataListMap.get(windowEnd);
dataList.sort((o1, o2) -> o2.f1 - o1.f1);
// 2. 取TopN
StringBuilder outStr = new StringBuilder();
outStr.append("================================\n");
// 遍历 排序后的 List,取出前 threshold 个, 考虑可能List不够2个的情况 ==》 List中元素的个数 和 2 取最小值
for (int i = 0; i < Math.min(threshold, dataList.size()); i++) {
Tuple3<Integer, Integer, Long> vcCount = dataList.get(i);
outStr.append("Top" + (i + 1) + "\n");
outStr.append("vc=" + vcCount.f0 + "\n");
outStr.append("count=" + vcCount.f1 + "\n");
outStr.append("窗口结束时间=" + vcCount.f2 + "\n");
outStr.append("================================\n");
}
out.collect(outStr.toString());
}
}
输入:
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
s1,2,1
s2,3,3
s1,4,3
s1,5,1
s2,8,1
s1,9,3
s3,11,3
s1,12,3
s1,13,3
输出:
================================
Top1
vc=1
count=2
窗口结束时间=5000
================================
Top2
vc=3
count=2
窗口结束时间=5000
================================
================================
Top1
vc=1
count=4
窗口结束时间=10000
================================
Top2
vc=3
count=3
窗口结束时间=10000
================================
3.3 侧输出流(Side Output)
侧输出流可以认为是“主流”上分叉出的“支流”,所以可以由一条流产生出多条流,而且这些流中的数据类型还可以不一样。利用这个功能可以很容易地实现“分流”操作。
具体实现上,可以在处理函数中的上下文中调用 .output() 方法即可。
案例:对每个传感器,水位超过10则输出告警信息。
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl())
.assignTimestampsAndWatermarks(WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L)
);
// 定义输出流标签
OutputTag<String> warnTag = new OutputTag<String>("warn-tag", Types.STRING);
SingleOutputStreamOperator<WaterSensor> process = sensorDS
.keyBy(WaterSensor::getId)
.process(new KeyedProcessFunction<String, WaterSensor, WaterSensor>() {
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, WaterSensor>.Context ctx, Collector<WaterSensor> out) throws Exception {
if (value.getVc() >= 10) {
// 使用侧输出流告警
ctx.output(warnTag, "当前水位线:" + value.getVc() + ",触发阈值10!");
}
out.collect(value);
}
});
// 输出主流
process.print("主流");
// 输出侧流
process.getSideOutput(warnTag).printToErr("侧流-Warn");
env.execute();
}
输入:
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,3
s2,5,7
s1,9,10
s1,5,9
s5,11,11
输出:
主流> WaterSensor{id='s1', ts=1, vc=3}
主流> WaterSensor{id='s2', ts=5, vc=7}
侧流-Warn> 当前水位线:10,触发阈值10!
主流> WaterSensor{id='s1', ts=9, vc=10}
主流> WaterSensor{id='s1', ts=5, vc=9}
侧流-Warn> 当前水位线:11,触发阈值10!
主流> WaterSensor{id='s5', ts=11, vc=11}