《Flink学习笔记》——第七章 处理函数

为了让代码有更强大的表现力和易用性，Flink 本身提供了多层 API

在更底层，我们可以不定义任何具体的算子（比如 map，filter，或者 window），而只是提炼出一个统一的“处理”（process）操作——它是所有转换算子的一个概括性的表达，可以自定义处理逻辑，所以这一层接口就被叫作“处理函数”（process function）。是整个DataStream API的基础

7.1 基本处理函数

处理函数主要是定义数据流的转换操作，Flink提供的处理函数类接口ProcessFunction

7.1.1 处理函数的功能和使用

我们之前讲过的MapFunction（一一处理，仅仅拿到数据）、AggregateFunction（窗口聚合，除了数据还可以拿到当前的状态）

另外，RichMapFunction提供了获取上下文的方法——getRuntimeContext()，可以拿到状态，并行度、任务名等运行时信息

但上面这些无法拿到事件的时间戳或者当前水位线。

而在很多应用需求中，要求我们对时间有更精细的控制，需要能够获取水位线，甚至要“把控时间”、定义什么时候做什么事，这就不是基本的时间窗口能够实现的了，所以这个时候就要用到底层的API——处理函数ProcessFunction了

• 提供“定时服务”，可以通过它访问事件流中的事件、时间戳、水位线，甚至可以注册“定时事件”
• 继承了AbstractRichFunction，拥有富函数所有特性
• 可以直接将数据输出到侧输出流

使用：

​ 直接基于 DataStream 调用.process()方法就可以了。方法需要传入一个 ProcessFunction 作为参数，用来定义处理逻辑。

stream.process(new MyProcessFunction())

7.1.2 ProcessFunction解析

public abstract class ProcessFunction<I, O> extends AbstractRichFunction{public abstract void processElement(I var1, ProcessFunction<I, O>.Context var2, Collector<O> var3);public void onTimer(long timestamp, ProcessFunction<I, O>.OnTimerContext ctx, Collector<O> out);
}

1.抽象方法.processElement()

• var1：正在处理的数据
• var2：上下文
• var3：“收集器”，用于返回数据

2.非抽象方法.onTimer()

• 用于定义定时触发的操作

7.1.3 处理函数的分类

Flink 中的处理函数其实是一个大家族，ProcessFunction 只是其中一员

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）做连接之后的产物

7.2 按键分区处理函数

public abstract class KeyedProcessFunction<K, I, O> extends AbstractRichFunction

只有在 KeyedStream 中才支持使用 TimerService 设置定时器的操作，所以一般情况下，我们都是先做了 keyBy 分区之后，再去定义处理操作；代码中更加常见的处理函数是 KeyedProcessFunction，最基本的 ProcessFunction 反而出镜率没那么高。KeyedProcessFunction 可以说是处理函数中的“嫡系部队”，可以认为是 ProcessFunction 的一个扩展。

7.2.1 定时器（Timer）和定时服务（TimerService）

首先通过定时服务注册一个定时器，ProcessFunction 的上下文（Context）中提供了.timerService()方法，可以直接返回一个 TimerService 对象。

TimerService 是 Flink 关于时间和定时器的基础服务接口，包含以下六个方法：

// 获取当前的处理时间
long currentProcessingTime();// 获取当前的水位线（事件时间）
long currentWatermark();// 注册处理时间定时器，当处理时间超过 time 时触发
void registerProcessingTimeTimer(long time);// 注册事件时间定时器，当水位线超过 time 时触发
void registerEventTimeTimer(long time);// 删除触发时间为 time 的处理时间定时器
void deleteProcessingTimeTimer(long time);// 删除触发时间为 time 的处理时间定时器
void deleteEventTimeTimer(long time);

7.2.2 KeyedProcessFunction的使用

与 ProcessFunction 的定义几乎完全一样，区别只是在于类型参数多了一个 K， 这是当前按键分区的 key 的类型。在KeyedProcessFunction中可以注册定时器，定义定时器触发逻辑。

KeyedProcessFunction是个抽象类，继承了AbstractRichFunction。

public abstract class KeyedProcessFunction<K, I, O> extends AbstractRichFunction

主要有两个核心的方法：

// 定义处理每个元素的逻辑
public abstract void processElement(I value, Context ctx, Collector<O> out)// 定时器触发时处理逻辑
public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out)

从上面可以看到，参数里面都有Context（这里OnTimerContext继承了Context），所以都可以通过

ctx.timerService().registerEventTimeTimer(long time);

去注册定时器。

示例：

自定义数据源

public class CustomSource implements SourceFunction<Event> {@Overridepublic void run(SourceContext<Event> ctx) throws Exception {// 直接发出一条数据ctx.collect(new Event("Mark", "./hhhh.com", 1000L));// 中间停顿5秒Thread.sleep(5000L);// 发出10秒后的数据ctx.collect(new Event("Mark", "/home", 11000L));Thread.sleep(5000L);// 发出 10 秒+1ms 后的数据ctx.collect(new Event("Alice", "./cart", 11001L));Thread.sleep(5000L);}@Overridepublic void cancel() {}
}

创建一个KeyedProcessFunction实现类

public class MyKeyedProcessFunction extends KeyedProcessFunction<Boolean, Event, String> {@Overridepublic void processElement(Event value, KeyedProcessFunction<Boolean, Event, String>.Context ctx, Collector<String> out) throws Exception {out.collect("数据到达，时间戳为：" + ctx.timestamp());out.collect("数据到达，水位线为：" + ctx.timerService().currentWatermark());// 注册一个 1 秒后的定时器ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 1000L);out.collect(String.format("注册定时器：%d%n-------分割线-------", ctx.timestamp() + 1000L));}@Overridepublic void onTimer(long timestamp, KeyedProcessFunction<Boolean, Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {out.collect("定时器触发，触发时间：" + timestamp);}
}

主函数

public class EventTimeTimerTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<Event> stream = env.addSource(new CustomSource()).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<Event>() {@Overridepublic long extractTimestamp(Event element, long recordTimestamp) {return element.getTimestamp();}}));stream.keyBy(data->true).process(new MyKeyedProcessFunction()).print();env.execute();}
}

输出结果：

数据到达，时间戳为：1000
数据到达，水位线为：-9223372036854775808
注册定时器：2000
-------分割线-------
数据到达，时间戳为：11000
数据到达，水位线为：999
注册定时器：12000
-------分割线-------
定时器触发，触发时间：2000
数据到达，时间戳为：11001
数据到达，水位线为：10999
注册定时器：12001
-------分割线-------
定时器触发，触发时间：12000
定时器触发，触发时间：12001

输出结果解释：

当第一条数据 Event(“Mark”, “./hhhh.com”, 1000L) 过来，由于水位线生成的周期是默认（200ms）一次，所以第一次数据过来时，水位线没有更新，为默认值Long.MIN_VALUE，此时注册一个以事件时间为准加1000ms的定时器。所以输出就是：

数据到达，时间戳为：1000
数据到达，水位线为：-9223372036854775808
注册定时器：2000
-------分割线-------

过了200ms后，到了水位线生成时间，此时最大时间戳为1000，由于没有设置水位线延迟，所以默认减1ms。此时水位线为：1000-1=999.并未达到定时器触发时间（2000）

过了5秒钟第二条数据 Event(“Mark”, “/home”, 11000L) 过来，输出并注册了一个12000的定时器：

数据到达，时间戳为：11000
数据到达，水位线为：999
注册定时器：12000
-------分割线-------

达到水位线生成时间后，更新为11000-1=10999，此时达到（注册定时器：2000）触发时间，所以输出：

定时器触发，触发时间：2000

过了5秒，数据 Event(“Alice”, “./cart”, 11001L) 过来，输出并注册了一个12001的定时器

数据到达，时间戳为：11001
数据到达，水位线为：10999
注册定时器：12001
-------分割线-------

达到水位线生成时间后，更新为11001-1=11000

过了5秒，数据发送执行完毕，第三条数据发出后再过 5 秒，没有更多的数据生成了，整个程序运行结束将要退出，此时 Flink 会自动将水位线推进到长整型的最大值（Long.MAX_VALUE）。于是所有尚未触发的定时器这时就统一触发了，输出

定时器触发，触发时间：12000
定时器触发，触发时间：12001

7.3 窗口处理函数

除了按键分区的处理，还有就是窗口数据的处理，常用的有：

• ProcessWindowFunction
• ProcessAllWindowFunction

7.3.1 窗口处理函数的使用

进行窗口计算，我们可以直接调用现成的简单聚合方法（sum/max/min）,也可以通过调用.reduce()或.aggregate()来自定义一般的增量聚合函数（ReduceFunction/AggregateFucntion）；而对于更加复杂、需要窗口信息和额外状态的一些场景，我们还可以直接使用全窗口函数、把数据全部收集保存在窗口内，等到触发窗口计算时再统一处理。窗口处理函数就是一种典型的全窗口函数。

窗口处理函数 ProcessWindowFunction 的使用与其他窗口函数类似，也是基于WindowedStream 直接调用方法就可以，只不过这时调用的是.process()

stream.keyBy( t -> t.f0 ).window( TumblingEventTimeWindows.of(Time.seconds(10)) ).process(new MyProcessWindowFunction())

7.3.2 ProcessWindowFunction 解析

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window>extends AbstractRichFunction/*
IN: 输入数据类型
OUT：输出数据类型
KEY：数据中key的类型
W：窗口类型
*/

方法：

// 窗口数据的处理
public abstract void process(KEY key, Context context, Iterable<IN> elements, Collector<OUT> out);
/*
key: 键
context: 上下文
elements: 窗口收集到用来计算的所有数据，这是一个可迭代的集合类型
out: 发送输出结果的收集器
*/// 这主要是方便我们进行窗口的清理工作。如果我们自定义了窗口状态，那么必须在.clear()方法中进行显式地清除，避免内存溢出
public void clear(Context context);

还定义了一个抽象类

public abstract class Context implements java.io.Serializable
// 我们之前可以看到，processFunction用的都是Context，但是这里ProcessWindowFunction 自己定义了一个Context，他是没有定时器的。为什么呢？因为本身窗口操作已经起到了一个触发计算的时间点，一般情况下是没有必要去做定时操作的。如果非要这么做，可以使用窗口触发器Trigger，里面有一个TriggerContext

ProcessAllWindowFunction的用法相似

stream.windowAll( TumblingEventTimeWindows.of(Time.seconds(10)) ).process(new MyProcessAllWindowFunction())

7.4 应用案例——TopN

窗口的计算处理，在实际应用中非常常见。对于一些比较复杂的需求，如果增量聚合函数无法满足，我们就需要考虑使用窗口处理函数这样的“大招”了。

网站中一个非常经典的例子，就是实时统计一段时间内的热门 url。例如，需要统计最近

10 秒钟内最热门的两个 url 链接，并且每 5 秒钟更新一次。我们知道，这可以用一个滑动窗口来实现，而“热门度”一般可以直接用访问量来表示。于是就需要开滑动窗口收集 url 的访问数据，按照不同的 url 进行统计，而后汇总排序并最终输出前两名。这其实就是著名的“Top N” 问题。

很显然，简单的增量聚合可以得到 url 链接的访问量，但是后续的排序输出 Top N 就很难实现了。所以接下来我们用窗口处理函数进行实现。

7.4.1 使用 ProcessAllWindowFunction

一种最简单的想法是，我们干脆不区分 url 链接，而是将所有访问数据都收集起来，统一进行统计计算。所以可以不做 keyBy，直接基于 DataStream 开窗，然后使用全窗口函数ProcessAllWindowFunction 来进行处理。

在窗口中可以用一个 HashMap 来保存每个 url 的访问次数，只要遍历窗口中的所有数据， 自然就能得到所有 url 的热门度。最后把 HashMap 转成一个列表 ArrayList，然后进行排序、取出前两名输出就可以了

public class ProcessAllWindowTopN {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource()).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<Event>() {@Overridepublic long extractTimestamp(Event event, long l) {return event.getTimestamp();}}));SingleOutputStreamOperator<String> urlStream = stream.map(new MapFunction<Event, String>() {@Overridepublic String map(Event event) throws Exception {return event.getUrl();}});SingleOutputStreamOperator<String> result = urlStream.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))).process(new ProcessAllWindowFunction<String, String, TimeWindow>() {@Overridepublic void process(ProcessAllWindowFunction<String, String, TimeWindow>.Context context,Iterable<String> elements, Collector<String> out) throws Exception {HashMap<String, Long> urlCountMap = new HashMap<>(10);for (String url : elements) {if (urlCountMap.containsKey(url)) {long count = urlCountMap.get(url);urlCountMap.put(url, count + 1);} else {urlCountMap.put(url, 1L);}}// 转存为ArrayListArrayList<Tuple2<String, Long>> mapList = new ArrayList<Tuple2<String, Long>>();for (String key : urlCountMap.keySet()) {mapList.add(Tuple2.of(key, urlCountMap.get(key)));}mapList.sort(new Comparator<Tuple2<String, Long>>() {@Overridepublic int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {return o2.f1.intValue() - o1.f1.intValue();}});// 取排序后的前两名，构建输出结果StringBuilder result = new StringBuilder();result.append("========================================\n");for (int i = 0; i < 2; i++) {Tuple2<String, Long> temp = mapList.get(i);String info = "浏览量 No." + (i + 1) +"  url：" + temp.f0 +"  浏览量：" + temp.f1 +"  窗口结束时间：" + new Timestamp(context.window().getEnd()) + "\n";result.append(info);}result.append("========================================\n");out.collect(result.toString());}});result.print();env.execute();}
}

7.4.2 使用KeyedProcessFunction

直接将所有数据放在一个分区上进行开窗操作。这相当于将并行度强行设置为 1，在实际应用中是要尽量避免的。

思路：

（1）读取数据源

（2）提取时间戳并生成水位线

（3）按照url进行keyBy分区

（4）开长度为10s步长为5的滑动窗口

（5）使用增量聚合函数 AggregateFunction，并结合全窗口函数 WindowFunction 进行窗口聚合，得到每个 url、在每个统计窗口内的浏览量，包装成 UrlViewCount

（6）按照窗口进行 keyBy 分区操作

（7）对同一窗口的统计结果数据，使用 KeyedProcessFunction 进行收集并排序输出

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7iv6fbOt-1693232836517)(第七章处理函数.assets/image-20230406003916609.png)]

// 自定义增量聚合
public class UrlViewCountAgg implements AggregateFunction<Event, Long, Long> {@Overridepublic Long createAccumulator() {return 0L;}@Overridepublic Long add(Event event, Long accumulator) {return accumulator + 1;}@Overridepublic Long getResult(Long accumulator) {return accumulator;}@Overridepublic Long merge(Long aLong, Long acc1) {return null;}
}

便于按窗口统计

public class UrlViewCount {public String url;public Long count;public Long windowStart;public Long windowEnd;public UrlViewCount() {}public UrlViewCount(String url, Long count, Long windowStart, Long windowEnd) {this.url = url;this.count = count;this.windowStart = windowStart;this.windowEnd = windowEnd;}@Overridepublic String toString() {return "UrlViewCount{" +"url='" + url + '\'' +", count=" + count +", windowStart=" + windowStart +", windowEnd=" + windowEnd +'}';}
}

窗口聚合函数

public class UrlViewCountResult extends ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow> {@Overridepublic void process(String url, ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<UrlViewCount> out) throws Exception {long start = context.window().getStart();long end = context.window().getEnd();System.out.println(url);System.out.println(elements);out.collect(new UrlViewCount(url, elements.iterator().next(), start, end));}
}

排序取TopN

public class TopN extends KeyedProcessFunction<Long, UrlViewCount, String> {private Integer n;// 定义一个列表状态private ListState<UrlViewCount> urlViewCountListState;public TopN(Integer n) {this.n = n;}@Overridepublic void open(Configuration parameters) throws Exception {// 从环境中获取列表状态句柄urlViewCountListState = getRuntimeContext().getListState(new ListStateDescriptor<UrlViewCount>("url-view-count-list", Types.POJO(UrlViewCount.class)));}@Overridepublic void processElement(UrlViewCount value, KeyedProcessFunction<Long, UrlViewCount, String>.Context ctx, Collector<String> out) throws Exception {// 将 count 数据添加到列表状态中，保存起来urlViewCountListState.add(value);// 注册 window end + 1ms 后的定时器，等待所有数据到齐开始排序ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);}@Overridepublic void onTimer(long timestamp, KeyedProcessFunction<Long, UrlViewCount, String>.OnTimerContext ctx, Collector<String> out) throws Exception {// 将数据从列表状态变量中取出，放入 ArrayList，方便排序ArrayList<UrlViewCount> urlViewCountArrayList = new ArrayList<>();for (UrlViewCount urlViewCount : urlViewCountListState.get()) {urlViewCountArrayList.add(urlViewCount);}// 清空状态，释放资源urlViewCountListState.clear();// 排 序urlViewCountArrayList.sort(new Comparator<UrlViewCount>(){@Overridepublic int compare(UrlViewCount o1, UrlViewCount o2) {return o2.count.intValue() - o1.count.intValue();}});// 取前两名，构建输出结果StringBuilder result = new StringBuilder(); result.append("========================================\n");result.append("窗口结束时间：" + new Timestamp(timestamp - 1) + "\n");for (int i = 0; i < this.n; i++) {UrlViewCount UrlViewCount = urlViewCountArrayList.get(i); String info = "No." + (i + 1) + " "+ "url：" + UrlViewCount.url + " "+ "浏览量：" + UrlViewCount.count + "\n"; result.append(info);} result.append("========================================\n");out.collect(result.toString());}
}

主方法：

public class KeyedProcessTopN {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);// 从自定义数据源读取数据SingleOutputStreamOperator<Event> eventStream	=	env.addSource(new ClickSource()).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<Event>() {@Overridepublic long extractTimestamp(Event element,	long recordTimestamp) {return element.getTimestamp();}}));SingleOutputStreamOperator<UrlViewCount> urlCountStream = eventStream.keyBy(Event::getUrl).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))).aggregate(new UrlViewCountAgg(), new UrlViewCountResult());SingleOutputStreamOperator<String> result = urlCountStream.keyBy(data -> data.windowEnd).process(new TopN(2));result.print();env.execute();}
}

其实这里面是可以优化的。每次其实是把所有url——count都会发过来，保存到一个列表状态中。虽然只是一个窗口的，但是如果数据量大的话还是可以优化的。

7.5 侧输出流

处理函数还有另外一个特有功能，就是将自定义的数据放入“侧输出流”（side output）输出。这个概念我们并不陌生，之前在讲到窗口处理迟到数据时，最后一招就是输出到侧输出流。而这种处理方式的本质，其实就是处理函数的侧输出流功能。

我们之前讲到的绝大多数转换算子，输出的都是单一流，流里的数据类型只能有一种。而侧输出流可以认为是“主流”上分叉出的“支流”，所以可以由一条流产生出多条流，而且这些流中的数据类型还可以不一样。利用这个功能可以很容易地实现“分流”操作。

具体应用时，只要在处理函数的.processElement()或者.onTimer()方法中，调用上下文的.output()方法就可以了

DataStream<Integer> stream = env.addSource(...);
SingleOutputStreamOperator<Long> process = eventStream.process(new ProcessFunction<Integer, Long>() {@Overridepublic void processElement(Integer value, ProcessFunction<Integer, Long>.Context ctx, Collector<Long> out) throws Exception {out.collect(Long.valueOf(value));ctx.output(outputTag, "side-output: " + value);}
});

这里 output()方法需要传入两个参数，第一个是一个“输出标签”OutputTag，用来标识侧输出流，一般会在外部统一声明；第二个就是要输出的数据。

我们可以在外部先将 OutputTag 声明出来

OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

如果想要获取这个侧输出流，可以基于处理之后的 DataStream 直接调用.getSideOutput() 方法，传入对应的 OutputTag，这个方式与窗口 API 中获取侧输出流是完全一样的。

DataStream<String> stringStream = longStream.getSideOutput(outputTag);