Flink从入门到上天系列第十四篇：Flink当中的处理函数

一：Flink当中的处理函数

之前所介绍的流处理API，无论是基本的转换、聚合，还是更为复杂的窗口操作，其实都是基于DataStream进行转换的，所以可以统称为DataStream API。

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

这个功能就是底层API。自己干就好了。当我们不知道到底要使用什么算子的时候，我们直接使用process就好了。process可以多种内容。

二：基本处理函数

1：处理函数的功能和使用

Flink 的处理函数是实现精细时间控制、自定义业务逻辑的底层算子，相比普通转换算子更灵活：

1. 核心能力：可通过内置的定时服务（TimerService）访问事件时间戳、水位线，还能注册定时事件；
2. 基础特性：继承 AbstractRichFunction，支持访问状态、运行时信息，也可输出数据到侧输出流；
3. 使用方式：通过 DataStream 调用.process()方法，传入 ProcessFunction 抽象类的自定义实现类即可。

关键点回顾

1. 处理函数解决了普通转换算子无法访问时间戳、水位线的问题，支持精细时间控制；
2. 兼具富函数（RichFunction）的所有特性，可访问状态、输出到侧输出流；
3. 用法简单，通过.process()+ 自定义 ProcessFunction 实现自定义逻辑。

2：ProcessFunction的解析

在源码中我们可以看到，抽象类ProcessFunction继承了AbstractRichFunction，有两个泛型类型参数：I表示Input，也就是输入的数据类型；O表示Output，也就是处理完成之后输出的数据类型。

内部单独定义了两个方法：一个是必须要实现的抽象方法.processElement()；另一个是非抽象方法.onTimer()。

@PublicEvolving public abstract class ProcessFunction<I, O> extends AbstractRichFunction { private static final long serialVersionUID = 1L; 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 {} public abstract class Context { public abstract Long timestamp(); public abstract TimerService timerService(); public abstract <X> void output(OutputTag<X> outputTag, X value); } public abstract class OnTimerContext extends Context { public abstract TimeDomain timeDomain(); } }

1：抽象方法.processElement()

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

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

通过几个参数的分析不难发现，ProcessFunction可以轻松实现flatMap、map、filter这样的基本转换功能；而通过富函数提供的获取上下文方法.getRuntimeContext()，也可以自定义状态（state）进行处理，这也就能实现聚合操作的功能了。

2）非抽象方法.onTimer()

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

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

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

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

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

4：总结

这些函数的，都是使用匿名内部类的方式给填充到process函数当中。

三：总结

Flink API双流Join的两种实现方式

1：基于窗口的实现

flink: 双流Join: 实现方式1: 基于窗口的Join 核心API调用顺序: - ds1.join(ds2) # 关联两个数据流 - where(键提取逻辑) # 提取第一个流的关联键 - equalTo(键提取逻辑) # 提取第二个流的关联键 - window(窗口类型) # 定义关联的窗口范围（如滚动/滑动窗口） - apply(Join处理逻辑) # 自定义关联后的处理逻辑 实现方式2: 基于状态的Interval Join 核心API: intervalJoin() # 基于状态和时间区间的关联方式 核心特点: 无需显式定义传统窗口，通过时间区间匹配双流数据

2：基于状态的实现

flink: 双流Join: 实现方式1: 基于窗口的Join 核心API: ds1.join(ds2).where().equalTo().window().apply() 实现方式2: 基于状态的Interval Join 核心API: - keyedStream.intervalJoin(keyedStream) - between(下界, 上界) - process() 底层原理: - 数据流存入状态缓存 - 用当前数据匹配另一流缓存数据 - 超时数据清理状态 注意: Flink API仅支持内连接；外连接需用Flink SQL或connect实现

3：处理函数总结

处理函数: 底层: 最底层API，可精细控制时间与状态 继承: AbstractRichFunction，具备富函数能力 核心方法: processElement: 处理单条数据 onTimer: 定时器触发回调 能力: 访问时间戳、水位线、注册定时器、输出侧输出流