Flink 中的处理函数

1.概述

对于不同类型的流，都可以直接调用 process() 方法进行自定义处理，这时传入的参数叫作处理函数。

处理函数提供了一个“定时服务”（TimerService），可以通过它访问流中的事件（event）、时间戳（timestamp）、水位线（watermark），甚至可以注册“定时事件”。而且处理函数继承了AbstractRichFunction抽象类，所以拥有富函数类的所有特性，同样可以访问状态（state）和其他运行时信息。此外，处理函数还可以直接将数据输出到侧输出流（side output）中。

2. 基本处理函数（ProcessFunction）

public abstract class ProcessFunction<I, O> extends AbstractRichFunction {

    public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception{}

    public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) throws Exception {}

}

1）processElement()

用于“处理元素”，定义了处理的核心逻辑。这个方法对于流中的每个元素都会调用一次，参数包括三个：输入数据值value，上下文ctx，以及“收集器”（Collector）out。方法没有返回值，处理之后的输出数据是通过收集器out来定义的。

• value：当前流中的输入元素，也就是正在处理的数据，类型与流中数据类型一致。
• ctx：类型是 ProcessFunction 中定义的内部抽象类 Context，表示当前运行的上下文，可以获取到当前的时间戳，并提供了用于查询时间和注册定时器的“定时服务”（TimerService），以及可以将数据发送到“侧输出流”的方法.output()。
• out：“收集器”（类型为Collector），用于返回输出数据。使用方式与flatMap算子中的收集器完全一样，直接调用 out.collect() 方法就可以向下游发出一个数据。

2）onTimer()

这个方法只有在注册好的定时器触发的时候才会调用，而定时器是通过“定时服务” TimerService 来注册的。打个比方，注册定时器（timer）就是设了一个闹钟，到了设定时间就会响；而 onTimer()中定义的就是闹钟响的时候要做的事。所以它本质上是一个基于时间的“回调”（callback）方法，通过时间的进展来触发；在事件时间语义下就是由水位线（watermark）来触发了。

定时方法 onTimer() 有三个参数：时间戳（timestamp），上下文（ctx），以及收集器（out）。这里的 timestamp 是指设定好的触发时间，事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器，所以也可以调用定时服务（TimerService），以及任意输出处理之后的数据。

onTimer()方法做定时触发，ProcessFunction也可以自定义数据按照时间分组、定时触发计算输出结果；这其实就实现了窗口（window）的功能。

注意：在Flink中，只有“按键分组流” KeyedStream 才支持设置定时器的操作。

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

onTimer() 方法中可以实现定时处理的逻辑，而它能触发的前提，就是之前曾经注册过定时器、并且现在已经到了触发时间。注册定时器的功能，是通过上下文中提供的“定时服务”来实现的。

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

注意：TimerService 会以键（key）和时间戳为标准，对定时器进行去重；也就是说对于每个key和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。

3. 处理函数的分类

对于不同类型的流，都可以直接调用 process() 方法进行自定义处理，这时传入的参数就都叫作处理函数。

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

4. 侧输出流

1）在处理函数的 processElement() 或者 onTimer() 方法中，调用上下文的 output() 方法就可以将数据放入侧输出流。

2）基于 DataStream 调用 getSideOutput() 方法。