[Flink]二、Flink1.13

7. 处理函数

之前所介绍的流处理 API，无论是基本的转换、聚合，还是更为复杂的窗口操作，其实都是基于 DataStream 进行转换的；所以可以统称为 DataStream API ，这也是 Flink 编程的核心。而我们知道，为了让代码有更强大的表现力和易用性， Flink 本身提供了多层 API ，DataStream API 只是中间的一环，如图 7-1 所示：

在更底层，我们可以不定义任何具体的算子（比如 map ， filter ，或者 window ），而只是提

炼出一个统一的“处理”（ process ）操作——它是所有转换算子的一个概括性的表达，可以自

定义处理逻辑，所以这一层接口就被叫作“处理函数”（ process function ）。

在处理函数中，我们直面的就是数据流中最基本的元素：数据事件（event ）、状态（ state ）

以及时间（ time ）。这就相当于对流有了完全的控制权。处理函数比较抽象，没有具体的操作，

所以对于一些常见的简单应用（比如求和、开窗口）会显得有些麻烦；不过正是因为它不限定

具体做什么，所以理论上我们可以做任何事情，实现所有需求。所以可以说，处理函数是我们

进行 Flink 编程的“大招”，轻易不用，一旦放出来必然会扫平一切。

本章我们就深入底层，讨论一下 Flink 中处理函数的使用方法。

7.1 基本处理函数（ ProcessFunction ）

处理函数主要是定义数据流的转换操作，所以也可以把它归到转换算子中。我们知道在

Flink 中几乎所有转换算子都提供了对应的函数类接口，处理函数也不例外；它所对应的函数

类，就叫作 ProcessFunction 。

7.1.1 处理函数的功能和使用

我们之前学习的转换算子，一般只是针对某种具体操作来定义的，能够拿到的信息比较有

限。比如 map 算子，我们实现的 MapFunction 中，只能获取到当前的数据，定义它转换之后

的形式；而像窗口聚合这样的复杂操作， AggregateFunction 中除数据外，还可以获取到当前的

183 状态（以累加器 Accumulator 形式出现）。另外我们还介绍过富函数类，比如 RichMapFunction ，

它提供了获取运行时上下文的方法 getRuntimeContext() ，可以拿到状态，还有并行度、任务名

称之类的运行时信息。

但是无论那种算子，如果我们想要访问事件的时间戳，或者当前的水位线信息，都是完全

做不到的。在定义生成规则之后，水位线会源源不断地产生，像数据一样在任务间流动，可我

们却不能像数据一样去处理它；跟时间相关的操作，目前我们只会用窗口来处理。而在很多应

用需求中，要求我们对时间有更精细的控制，需要能够获取水位线，甚至要 “ 把控时间 ” 、定义

什么时候做什么事，这就不是基本的时间窗口能够实现的了。

于是必须祭出大招——处理函数（ ProcessFunction ）了。处理函数提供了一个“定时服务”

（ TimerService ），我们可以通过它访问流中的事件（ event ）、时间戳（ timestamp ）、水位线

（ watermark ），甚至可以注册 “ 定时事件 ” 。而且处理函数继承了 AbstractRichFunction 抽象类，

所以拥有富函数类的所有特性，同样可以访问状态（ state ）和其他运行时信息。此外，处理函

数还可以直接将数据输出到侧输出流（ side output ）中。所以，处理函数是最为灵活的处理方

法，可以实现各种自定义的业务逻辑；同时也是整个 DataStream API 的底层基础。

处理函数的使用与基本的转换操作类似，只需要直接基于 DataStream 调用 .process() 方法

就可以了。方法需要传入一个 ProcessFunction 作为参数，用来定义处理逻辑。

stream.process(new MyProcessFunction())

这里 ProcessFunction 不是接口，而是一个抽象类，继承了 AbstractRichFunction ；

MyProcessFunction 是它的一个具体实现。所以所有的处理函数，都是富函数（ RichFunction ），

富函数可以调用的东西这里同样都可以调用。

下面是一个具体的应用示例：

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
public class ProcessFunctionExample {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 env
 .addSource(new ClickSource())
 .assignTimestampsAndWatermarks(
 WatermarkStrategy.<Event>forMonotonousTimestamps()
 .withTimestampAssigner(new 
SerializableTimestampAssigner<Event>() {
 @Override
 public long extractTimestamp(Event event, long l) {
 return event.timestamp;
 }
 })
 )
 .process(new ProcessFunction<Event, String>() {
 @Override
 public void processElement(Event value, Context ctx, 
Collector<String> out) throws Exception {
 if (value.user.equals("Mary")) {
 out.collect(value.user);
 } else if (value.user.equals("Bob")) {
 out.collect(value.user);
 out.collect(value.user);
 }
 System.out.println(ctx.timerService().currentWatermark());
 }
 })
 .print();
 env.execute();
 }
}

这里我们在 ProcessFunction 中重写了 .processElement() 方法，自定义了一种处理逻辑：当

数据的 user 为“ Mary ”时，将其输出一次；而如果为“ Bob ”时，将 user 输出两次。这里的

输出，是通过调用 out.collect() 来实现的。另外我们还可以调用

ctx.timerService().currentWatermark() 来获取当前的水位线打印输出。所以可以看到，

ProcessFunction 函数有点像 FlatMapFunction 的升级版。可以实现 Map 、 Filter 、 FlatMap 的所

有功能。很明显，处理函数非常强大，能够做很多之前做不到的事情。

接下来我们就深入 ProcessFunction 内部来进行详细了解。

7.1.2 ProcessFunction 解析

在源码中我们可以看到，抽象类 ProcessFunction 继承了 AbstractRichFunction ，有两个泛

型类型参数： I 表示 Input ，也就是输入的数据类型； O 表示 Output ，也就是处理完成之后输出

的数据类型。

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

抽象方法 .onTimer() 。

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) ，以及可以将数据发送到“侧输出流”（ side output ）的方法 .output() 。

Context 抽象类定义如下：

public abstract class Context {

public abstract Long timestamp();

public abstract TimerService timerService();

public abstract <X> void output(OutputTag<X> outputTag, X value);

}

⚫ out ：“收集器”（类型为 Collector ），用于返回输出数据。使用方式与 flatMap

算子中的收集器完全一样，直接调用 out.collect() 方法就可以向下游发出一个数据。

这个方法可以多次调用，也可以不调用。

通过几个参数的分析不难发现， ProcessFunction 可以轻松实现 flatMap 这样的基本转换功

能（当然 map 、 filter 更不在话下）；而通过富函数提供的获取上下文方法 .getRuntimeContext() ，

也可以自定义状态（ state ）进行处理，这也就能实现聚合操作的功能了。关于自定义状态的具

体实现，我们会在后续“状态管理”一章中详细介绍。

2. 非抽象方法 .onTimer()

用于定义定时触发的操作，这是一个非常强大、也非常有趣的功能。这个方法只有在注册

好的定时器触发的时候才会调用，而定时器是通过“定时服务” TimerService 来注册的。打个

比方，注册定时器（ timer ）就是设了一个闹钟，到了设定时间就会响；而 .onTimer() 中定义的，

就是闹钟响的时候要做的事。所以它本质上是一个基于时间的“回调”（ callback ）方法，通过

时间的进展来触发；在事件时间语义下就是由水位线（ watermark ）来触发了。

与 .processElement() 类似，定时方法 .onTimer() 也有三个参数：时间戳（ timestamp ），上下

文（ ctx ），以及收集器（ out ）。这里的 timestamp 是指设定好的触发时间，事件时间语义下当

186 然就是水位线了。另外这里同样有上下文和收集器，所以也可以调用定时服务（ TimerService ），

以及任意输出处理之后的数据。

既然有 .onTimer() 方法做定时触发，我们用 ProcessFunction 也可以自定义数据按照时间分

组、定时触发计算输出结果；这其实就实现了窗口（ window ）的功能。所以说 ProcessFunction

是真正意义上的终极奥义，用它可以实现一切功能。

我们也可以看到，处理函数都是基于事件触发的。水位线就如同插入流中的一条数据一样；

只不过处理真正的数据事件调用的是 .processElement() 方法，而处理水位线事件调用的

是 .onTimer() 。

这里需要注意的是，上面的 .onTimer() 方法只是定时器触发时的操作，而定时器（ timer ）

真正的设置需要用到上下文 ctx 中的定时服务。在 Flink 中，只有“按键分区流” KeyedStream

才支持设置定时器的操作，所以之前的代码中我们并没有使用定时器。所以基于不同类型的流，

可以使用不同的处理函数，它们之间还是有一些微小的区别的。接下来我们就介绍一下处理函

数的分类。

7.1.3 处理函数的分类

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

我们知道， 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

187 合并（ 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 和 ProcessWindowFunction 的具体用法展开详细

说明。

7.2 按键分区处理函数（ KeyedProcessFunction ）

在 Flink 程序中，为了实现数据的聚合统计，或者开窗计算之类的功能，我们一般都要先

用 keyBy 算子对数据流进行“按键分区”，得到一个 KeyedStream 。也就是指定一个键（ key ），

按照它的哈希值（ hash code ）将数据分成不同的“组”，然后分配到不同的并行子任务上执行

计算；这相当于做了一个逻辑分流的操作，从而可以充分利用并行计算的优势实时处理海量数

据。

另外我们在上节中也提到，只有在 KeyedStream 中才支持使用 TimerService 设置定时器的

操作。所以一般情况下，我们都是先做了 keyBy 分区之后，再去定义处理操作；代码中更加

常见的处理函数是 KeyedProcessFunction ，最基本的 ProcessFunction 反而出镜率没那么高。

接下来我们就先从定时服务（ TimerService ）入手，详细讲解 KeyedProcessFunction 的用

法

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

KeyedProcessFunction 的一个特色，就是可以灵活地使用定时器。

定时器（ timers ）是处理函数中进行时间相关操作的主要机制。在 .onTimer() 方法中可以实

现定时处理的逻辑，而它能触发的前提，就是之前曾经注册过定时器、并且现在已经到了触发

188 时间。注册定时器的功能，是通过上下文中提供的“定时服务”（ TimerService ）来实现的。

定时服务与当前运行的环境有关。前面已经介绍过， ProcessFunction 的上下文（ Context ）

中提供了 .timerService() 方法，可以直接返回一个 TimerService 对象：

public abstract 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 ，不过只有基于 KeyedStream 的处理函数，才能去调用注册和删除定时器的方法；

未作按键分区的 DataStream 不支持定时器操作，只能获取当前时间。

对于处理时间和事件时间这两种类型的定时器， TimerService 内部会用一个优先队列将它

们的时间戳（ timestamp ）保存起来，排队等待执行。可以认为，定时器其实是 KeyedStream

上处理算子的一个状态，它以时间戳作为区分。所以 TimerService 会以键（ key ）和时间戳为

标准，对定时器进行去重；也就是说对于每个 key 和时间戳，最多只有一个定时器，如果注册

了多次， onTimer() 方法也将只被调用一次。这样一来，我们在代码中就方便了很多，可以肆

无忌惮地对一个 key 注册定时器，而不用担心重复定义——因为一个时间戳上的定时器只会触

发一次。

基于 KeyedStream 注册定时器时，会传入一个定时器触发的时间戳，这个时间戳的定时器

对于每个 key 都是有效的。这样，我们的代码并不需要做额外的处理，底层就可以直接对不同

key 进行独立的处理操作了。

利用这个特性，有时我们可以故意降低时间戳的精度，来减少定时器的数量，从而提高处

理性能。比如我们可以在设置定时器时只保留整秒数，那么定时器的触发频率就是最多 1 秒一

次。

long coalescedTime = time / 1000 * 1000;

189 ctx.timerService().registerProcessingTimeTimer(coalescedTime);

这里注意定时器的时间戳必须是毫秒数，所以我们得到整秒之后还要乘以 1000 。定时器

默认的区分精度是毫秒。

另外 Flink 对 .onTimer() 和 .processElement() 方法是同步调用的（ synchronous ），所以也不会

出现状态的并发修改。

Flink 的定时器同样具有容错性，它和状态一起都会被保存到一致性检查点（ checkpoint ）

中。当发生故障时， Flink 会重启并读取检查点中的状态，恢复定时器。如果是处理时间的定

时器，有可能会出现已经“过期”的情况，这时它们会在重启时被立刻触发。关于 Flink 的检

查点和容错机制，我们会在后续章节详细讲解。

7.2.2 KeyedProcessFunction 的使用

KeyedProcessFunction 可以说是处理函数中的 “ 嫡系部队 ” ，可以认为是 ProcessFunction 的

一个扩展。我们只要基于 keyBy 之后的 KeyedStream ，直接调用 .process() 方法，这时需要传入

的参数就是 KeyedProcessFunction 的实现类。

stream.keyBy( t -> t.f0 )

.process(new MyKeyedProcessFunction())

类似地， KeyedProcessFunction 也是继承自 AbstractRichFunction 的一个抽象类，源码中定

义如下：

public abstract class KeyedProcessFunction<K, 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 {}

public abstract class Context {...}

...

}

可以看到与 ProcessFunction 的定义几乎完全一样，区别只是在于类型参数多了一个 K ，

这是当前按键分区的 key 的类型。同样地，我们必须实现一个 .processElement() 抽象方法，用

来处理流中的每一个数据；另外还有一个非抽象方法 .onTimer() ，用来定义定时器触发时的回

调操作。由于定时器只能在 KeyedStream 上使用，所以到了 KeyedProcessFunction 这里，我们

才真正对时间有了精细的控制，定时方法 .onTimer() 才真正派上了用场。

下面是一个使用处理时间定时器的具体示例：

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.sql.Timestamp;
public class ProcessingTimeTimerTest {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 // 处理时间语义，不需要分配时间戳和 watermark
 SingleOutputStreamOperator<Event> stream = env.addSource(new 
ClickSource());
 // 要用定时器，必须基于 KeyedStream
 stream.keyBy(data -> true)
 .process(new KeyedProcessFunction<Boolean, Event, String>() {
 @Override
 public void processElement(Event value, Context ctx, 
Collector<String> out) throws Exception {
 Long currTs = ctx.timerService().currentProcessingTime();
 out.collect("数据到达，到达时间：" + new Timestamp(currTs));
 // 注册一个 10 秒后的定时器
 ctx.timerService().registerProcessingTimeTimer(currTs + 10 
* 1000L);
 }
 @Override
 public void onTimer(long timestamp, OnTimerContext ctx, 
Collector<String> out) throws Exception {
 out.collect("定时器触发，触发时间：" + new Timestamp(timestamp));
 }
 })
 .print();
 env.execute();
 }
}

在上面的代码中，由于定时器只能在 KeyedStream 上使用，所以先要进行 keyBy ；这里

的 .keyBy(data -> true) 是将所有数据的 key 都指定为了 true ，其实就是所有数据拥有相同的 key ，

会分配到同一个分区。

之后我们自定义了一个 KeyedProcessFunction ，其中 .processElement() 方法是每来一个数据

都会调用一次，主要是定义了一个 10 秒之后的定时器；而 .onTimer() 方法则会在定时器触发时

调用。所以我们会看到，程序运行后先在控制台输出“数据到达”的信息，等待 10 秒之后，

又会输出“定时器触发”的信息，打印出的时间间隔正是 10 秒。

当然，上面的例子是处理时间的定时器，所以我们是真的需要等待 10 秒才会看到结果。

事件时间语义下，又会有什么不同呢？我们可以对上面的代码略作修改，做一个测试：

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
public class EventTimeTimerTest {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
192
 env.setParallelism(1);
 SingleOutputStreamOperator<Event> stream = env.addSource(new 
CustomSource())
 .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonot
onousTimestamps()
 .withTimestampAssigner(new 
SerializableTimestampAssigner<Event>() {
 @Override
 public long extractTimestamp(Event element, long 
recordTimestamp) {
 return element.timestamp;
 }
 }));
 // 基于 KeyedStream 定义事件时间定时器
 stream.keyBy(data -> true)
 .process(new KeyedProcessFunction<Boolean, Event, String>() {
 @Override
 public void processElement(Event value, Context ctx, 
Collector<String> out) throws Exception {
 out.collect("数据到达，时间戳为：" + ctx.timestamp());
 out.collect(" 数据到达，水位线为： " + 
ctx.timerService().currentWatermark() + "\n -------分割线-------");
 // 注册一个 10 秒后的定时器
 ctx.timerService().registerEventTimeTimer(ctx.timestamp() 
+ 10 * 1000L);
 }
 @Override
 public void onTimer(long timestamp, OnTimerContext ctx, 
Collector<String> out) throws Exception {
 out.collect("定时器触发，触发时间：" + timestamp);
 }
 })
 .print();
 env.execute();
 }
 // 自定义测试数据源
 public static class CustomSource implements SourceFunction<Event> {
 @Override
 public void run(SourceContext<Event> ctx) throws Exception {
 // 直接发出测试数据
 ctx.collect(new Event("Mary", "./home", 1000L));
 // 为了更加明显，中间停顿 5 秒钟
 Thread.sleep(5000L);
 // 发出 10 秒后的数据
 ctx.collect(new Event("Mary", "./home", 11000L));
 Thread.sleep(5000L);
 // 发出 10 秒+1ms 后的数据
 ctx.collect(new Event("Alice", "./cart", 11001L));
 Thread.sleep(5000L);
 }
 @Override
 public void cancel() { }
 }
}

由于是事件时间语义，所以我们必须从数据中提取出数据产生的时间戳。这里为了更清楚

地看到程序行为，我们自定义了一个数据源，发出三条测试数据，时间戳分别为 1000 、 11000

和 11001 ，并且发出数据后都会停顿 5 秒。

在代码中，我们依然将所有数据分到同一分区，然后在自定义的 KeyedProcessFunction 中

使用定时器。同样地，每来一条数据，我们就将当前的数据时间戳和水位线信息输出，并注册

一个 10 秒后（以当前数据时间戳为基准）的事件时间定时器。执行程序结果如下：

每来一条数据，都会输出两行“数据到达”的信息，并以分割线隔开；两条数据到达的时

间间隔为 5 秒。当第三条数据到达后，随后立即输出一条定时器触发的信息；再过 5 秒之后，

剩余两条定时器信息输出，程序运行结束。

我们可以发现，数据到来之后，当前的水位线与时间戳并不是一致的。当第一条数据到来，

时间戳为 1000 ，可水位线的生成是周期性的（默认 200ms 一次），不会立即发生改变，所以依

然是最小值 Long.MIN_VALUE ；随后只要到了水位线生成的时间点（ 200ms 到了），就会依据

当前的最大时间戳 1000 来生成水位线了。这里我们没有设置水位线延迟，默认需要减去 1 毫

秒，所以水位线推进到了 999 。而当时间戳为 11000 的第二条数据到来之后，水位线同样没有

立即改变，仍然是 999 ，就好像总是“滞后”数据一样。

这样程序的行为就可以得到合理解释了。事件时间语义下，定时器触发的条件就是水位线

推进到设定的时间。第一条数据到来后，设定的定时器时间为 1000 + 10 * 1000 = 11000 ；而当

时间戳为 11000 的第二条数据到来，水位线还处在 999 的位置，当然不会立即触发定时器；而

之后水位线会推进到 10999 ，同样是无法触发定时器的。必须等到第三条数据到来，将水位线

真正推进到 11000 ，就可以触发第一个定时器了。第三条数据发出后再过 5 秒，没有更多的数

据生成了，整个程序运行结束将要退出，此时 Flink 会自动将水位线推进到长整型的最大值

（ Long.MAX_VALUE ）。于是所有尚未触发的定时器这时就统一触发了，我们就在控制台看到

了后两个定时器的触发信息。

7.3 窗口处理函数

除了 KeyedProcessFunction ，另外一大类常用的处理函数，就是基于窗口的

ProcessWindowFunction 和 ProcessAllWindowFunction 了。如果看了前面的章节，会发现我们

195 之前已经简单地使用过窗口处理函数了。

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 解析

ProcessWindowFunction 既是处理函数又是全窗口函数。从名字上也可以推测出，它的本

质似乎更倾向于“窗口函数”一些。事实上它的用法也确实跟其他处理函数有很大不同。我们

可以从源码中的定义看到这一点：

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window>

extends AbstractRichFunction {

...

public abstract void process(

KEY key, Context context, Iterable<IN> elements, Collector<OUT> out) throws

Exception;

public void clear(Context context) throws Exception {}

public abstract class Context implements java.io.Serializable {...}

}

ProcessWindowFunction 依然是一个继承了 AbstractRichFunction 的抽象类，它有四个类型

参数：

⚫ IN ： input ，数据流中窗口任务的输入数据类型。

⚫ OUT ： output ，窗口任务进行计算之后的输出数据类型。

⚫ KEY ：数据中键 key 的类型。

196 197

⚫ W ：窗口的类型，是 Window 的子类型。一般情况下我们定义时间窗口， W

就是 TimeWindow 。

而内部定义的方法，跟我们之前熟悉的处理函数就有所区别了。因为全窗口函数不是逐个

处理元素的，所以处理数据的方法在这里并不是 .processElement() ，而是改成了 .process() 。方

法包含四个参数。

⚫ key ：窗口做统计计算基于的键，也就是之前 keyBy 用来分区的字段。

⚫ context ：当前窗口进行计算的上下文，它的类型就是 ProcessWindowFunction

内部定义的抽象类 Context 。

⚫ elements ：窗口收集到用来计算的所有数据，这是一个可迭代的集合类型。

⚫ out ：用来发送数据输出计算结果的收集器，类型为 Collector 。

可以明显看出，这里的参数不再是一个输入数据，而是窗口中所有数据的集合。而上下文

context 所包含的内容也跟其他处理函数有所差别：

public abstract class Context implements java.io.Serializable {

public abstract W window();

public abstract long currentProcessingTime();

public abstract long currentWatermark();

public abstract KeyedStateStore windowState();

public abstract KeyedStateStore globalState();

public abstract <X> void output(OutputTag<X> outputTag, X value);

}

除了可以通过 .output() 方法定义侧输出流不变外，其他部分都有所变化。这里不再持有

TimerService 对象，只能通过 currentProcessingTime() 和 currentWatermark() 来获取当前时间，所

以失去了设置定时器的功能；另外由于当前不是只处理一个数据，所以也不再提供 .timestamp()

方法。与此同时，也增加了一些获取其他信息的方法：比如可以通过 .window() 直接获取到当

前的窗口对象，也可以通过 .windowState() 和 .globalState() 获取到当前自定义的窗口状态和全局

状态。注意这里的“窗口状态”是自定义的，不包括窗口本身已经有的状态，针对当前 key 、

当前窗口有效；而“全局状态”同样是自定义的状态，针对当前 key 的所有窗口有效。

所以我们会发现， ProcessWindowFunction 中除了 .process() 方法外，并没有 .onTimer() 方法，

而是多出了一个 .clear() 方法。从名字就可以看出，这主要是方便我们进行窗口的清理工作。如

果我们自定义了窗口状态，那么必须在 .clear() 方法中进行显式地清除，避免内存溢出。

这里有一个问题：没有了定时器，那窗口处理函数就失去了一个最给力的武器，如果我们

希望有一些定时操作又该怎么做呢？其实仔细思考会发现，对于窗口而言，它本身的定义就包

含了一个触发计算的时间点，其实一般情况下是没有必要再去做定时操作的。如果非要这么干，

Flink 也提供了另外的途径——使用窗口触发器（ Trigger ）。在触发器中也有一个 TriggerContext ，

它可以起到类似 TimerService 的作用：获取当前时间、注册和删除定时器，另外还可以获取当前的状态。这样设计无疑会让处理流程更加清晰——定时操作也是一种“触发”，所以我们就

让所有的触发操作归触发器管，而所有处理数据的操作则归窗口函数管。

至于另一种窗口处理函数 ProcessAllWindowFunction ，它的用法非常类似。区别在于它基

于的是 AllWindowedStream ，相当于对没有 keyBy 的数据流直接开窗并调用 .process() 方法 :

stream.windowAll( TumblingEventTimeWindows.of(Time.seconds(10)) )

.process(new MyProcessAllWindowFunction())

7.4 应用案例—— Top N

窗口的计算处理，在实际应用中非常常见。对于一些比较复杂的需求，如果增量聚合函数

无法满足，我们就需要考虑使用窗口处理函数这样的“大招”了。

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

10 秒钟内最热门的两个 url 链接，并且每 5 秒钟更新一次。我们知道，这可以用一个滑动窗口

来实现，而“热门度”一般可以直接用访问量来表示。于是就需要开滑动窗口收集 url 的访问

数据，按照不同的 url 进行统计，而后汇总排序并最终输出前两名。这其实就是著名的“ Top N ”

问题。

很显然，简单的增量聚合可以得到 url 链接的访问量，但是后续的排序输出 Top N 就很难

实现了。所以接下来我们用窗口处理函数进行实现。

7.4.1 使用 ProcessAllWindowFunction

一种最简单的想法是，我们干脆不区分 url 链接，而是将所有访问数据都收集起来，统一

进行统计计算。所以可以不做 keyBy ，直接基于 DataStream 开窗，然后使用全窗口函数

ProcessAllWindowFunction 来进行处理。

在窗口中可以用一个 HashMap 来保存每个 url 的访问次数，只要遍历窗口中的所有数据，

自然就能得到所有 url 的热门度。最后把 HashMap 转成一个列表 ArrayList ，然后进行排序、

取出前两名输出就可以了。

代码具体实现如下：

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import 
org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import 
198
org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
public class ProcessAllWindowTopN {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 
 SingleOutputStreamOperator<Event> eventStream = env.addSource(new 
ClickSource())
 .assignTimestampsAndWatermarks(
 WatermarkStrategy.<Event>forMonotonousTimestamps()
 .withTimestampAssigner(new 
SerializableTimestampAssigner<Event>() {
 @Override
 public long extractTimestamp(Event element, long 
recordTimestamp) {
 return element.timestamp;
 }
 })
 );
 // 只需要 url 就可以统计数量，所以转换成 String 直接开窗统计
 SingleOutputStreamOperator<String> result = eventStream
 .map(new MapFunction<Event, String>() {
 @Override
 public String map(Event value) throws Exception {
 return value.url;
 }
 })
199
200
 .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), 
Time.seconds(5))) // 开滑动窗口
 .process(new ProcessAllWindowFunction<String, String, TimeWindow>() 
{
 @Override
 public void process(Context context, Iterable<String> elements, 
Collector<String> out) throws Exception {
 HashMap<String, Long> urlCountMap = new HashMap<>();
 // 遍历窗口中数据，将浏览量保存到一个 HashMap 中
 for (String url : elements) {
 if (urlCountMap.containsKey(url)) {
 long count = urlCountMap.get(url);
 urlCountMap.put(url, count + 1L);
 } else {
 urlCountMap.put(url, 1L);
 }
 }
 ArrayList<Tuple2<String, Long>> mapList = new 
ArrayList<Tuple2<String, Long>>();
 // 将浏览量数据放入 ArrayList，进行排序
 for (String key : urlCountMap.keySet()) {
 mapList.add(Tuple2.of(key, urlCountMap.get(key)));
 }
 mapList.sort(new Comparator<Tuple2<String, Long>>() {
 @Override
 public 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 ，在实际应用中是要尽量避免的，所以 Flink 官

方也并不推荐使用 AllWindowedStream 进行处理。另外，我们在全窗口函数中定义了 HashMap

来统计 url 链接的浏览量，计算过程是要先收集齐所有数据、然后再逐一遍历更新 HashMap ，

这显然不够高效。如果我们可以利用增量聚合函数的特性，每来一条数据就更新一次对应 url

的浏览量，那么到窗口触发计算时只需要做排序输出就可以了。

基于这样的想法，我们可以从两个方面去做优化：一是对数据进行按键分区，分别统计浏

览量；二是进行增量聚合，得到结果最后再做排序输出。所以，我们可以使用增量聚合函数

AggregateFunction 进行浏览量的统计，然后结合 ProcessWindowFunction 排序输出来实现 Top N

的需求。

具体实现思路就是，先按照 url 对数据进行 keyBy 分区，然后开窗进行增量聚合。这里就

会发现一个问题：我们进行按键分区之后，窗口的计算就会只针对当前 key 有效了；也就是说，

每个窗口的统计结果中，只会有一个 url 的浏览量，这是无法直接用 ProcessWindowFunction

进行排序的。所以我们只能分成两步：先对每个 url 链接统计出浏览量，然后再将统计结果收

集起来，排序输出最终结果。因为最后的排序还是基于每个时间窗口的，所以为了让输出的统

计结果中包含窗口信息，我们可以借用第六章中定义的 POJO 类 UrlViewCount 来表示，它包

201 含了 url 、浏览量（ count ）以及窗口的起始结束时间。之后对 UrlViewCount 的处理，可以先按

窗口分区，然后用 KeyedProcessFunction 来实现。

总结处理流程如下：

（ 1 ）读取数据源；

（ 2 ）筛选浏览行为（ pv ）；

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

（ 4 ）按照 url 进行 keyBy 分区操作；

（ 5 ）开长度为 1 小时、步长为 5 分钟的事件时间滑动窗口；

（ 6 ）使用增量聚合函数 AggregateFunction ，并结合全窗口函数 WindowFunction 进行窗口

聚合，得到每个 url 、在每个统计窗口内的浏览量，包装成 UrlViewCount ；

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

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

糟糕的是，这里又会带来另一个问题。最后我们用 KeyedProcessFunction 来收集数据做排

序，这时面对的就是窗口聚合之后的数据流，而窗口已经不存在了；那到底什么时候会收集齐

所有数据呢？这问题听起来似乎有些没道理。我们统计浏览量的窗口已经关闭，就说明了当前

已经到了要输出结果的时候，直接输出不就行了吗？

没有这么简单。因为数据流中的元素是逐个到来的，所以即使理论上我们应该“同时”收

到很多 url 的浏览量统计结果，实际也是有先后的、只能一条一条处理。下游任务（就是我们

定义的 KeyedProcessFunction ）看到一个 url 的统计结果，并不能保证这个时间段的统计数据

不会再来了，所以也不能贸然进行排序输出。解决的办法，自然就是要等所有数据到齐了——

这很容易让我们联想起水位线设置延迟时间的方法。这里我们也可以“多等一会儿”，等到水

位线真正超过了窗口结束时间，要统计的数据就肯定到齐了。

具体实现上，可以采用一个延迟触发的事件时间定时器。基于窗口的结束时间来设定延迟，

其实并不需要等太久——因为我们是靠水位线的推进来触发定时器，而水位线的含义就是“之

前的数据都到齐了”。所以我们只需要设置 1 毫秒的延迟，就一定可以保证这一点。

而在等待过程中，之前已经到达的数据应该缓存起来，我们这里用一个自定义的“列表状

态”（ ListState ）来进行存储，如图 7-2 所示。这个状态需要使用富函数类的 getRuntimeContext()

方法获取运行时上下文来定义，我们一般把它放在 open() 生命周期方法中。之后每来一个

UrlViewCount ，就把它添加到当前的列表状态中，并注册一个触发时间为窗口结束时间加 1

毫秒（ windowEnd + 1 ）的定时器。待到水位线到达这个时间，定时器触发，我们可以保证当

前窗口所有 url 的统计结果 UrlViewCount 都到齐了；于是从状态中取出进行排序输出。

具体代码实现如下：

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import 
org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import 
org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
public class KeyedProcessTopN {
 
public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 // 从自定义数据源读取数据
 SingleOutputStreamOperator<Event> eventStream = env.addSource(new 
ClickSource())
 .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonot
onousTimestamps()
 .withTimestampAssigner(new 
SerializableTimestampAssigner<Event>() {
 @Override
 public long extractTimestamp(Event element, long 
recordTimestamp) {
 return element.timestamp;
 }
 }));
 // 需要按照 url 分组，求出每个 url 的访问量
 SingleOutputStreamOperator<UrlViewCount> urlCountStream =
 eventStream.keyBy(data -> data.url)
 .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("result");
 env.execute();
 }
 // 自定义增量聚合
 public static class UrlViewCountAgg implements AggregateFunction<Event, Long, 
204
Long> {
 @Override
 public Long createAccumulator() {
 return 0L;
 }
 @Override
 public Long add(Event value, Long accumulator) {
 return accumulator + 1;
 }
 @Override
 public Long getResult(Long accumulator) {
 return accumulator;
 }
 @Override
 public Long merge(Long a, Long b) {
 return null;
 }
 }
 // 自定义全窗口函数，只需要包装窗口信息
 public static class UrlViewCountResult extends ProcessWindowFunction<Long, 
UrlViewCount, String, TimeWindow> {
 @Override
 public void process(String url, Context context, Iterable<Long> elements, 
Collector<UrlViewCount> out) throws Exception {
 // 结合窗口信息，包装输出内容
 Long start = context.window().getStart();
 Long end = context.window().getEnd();
 out.collect(new UrlViewCount(url, elements.iterator().next(), start, 
end));
 }
 }
 // 自定义处理函数，排序取 top n
 public static class TopN extends KeyedProcessFunction<Long, UrlViewCount, 
String>{
205
 // 将 n 作为属性
 private Integer n;
 // 定义一个列表状态
 private ListState<UrlViewCount> urlViewCountListState;
 public TopN(Integer n) {
 this.n = n;
 }
 @Override
 public void open(Configuration parameters) throws Exception {
 // 从环境中获取列表状态句柄
 urlViewCountListState = getRuntimeContext().getListState(
 new ListStateDescriptor<UrlViewCount>("url-view-count-list",
 Types.POJO(UrlViewCount.class)));
 }
 @Override
 public void processElement(UrlViewCount value, Context ctx, 
Collector<String> out) throws Exception {
 // 将 count 数据添加到列表状态中，保存起来
 urlViewCountListState.add(value);
 // 注册 window end + 1ms 后的定时器，等待所有数据到齐开始排序
 ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);
 }
 @Override
 public void onTimer(long timestamp, 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>() {
 @Override
206
207
 public 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());
 }
 }
}

代码中，我们还利用了定时器的特性：针对同一 key 、同一时间戳会进行去重。所以对于

同一个窗口而言，我们接到统计结果数据后设定的 windowEnd + 1 的定时器都是一样的，最终

只会触发一次计算。而对于不同的 key （这里 key 是 windowEnd ），定时器和状态都是独立的，

所以我们也不用担心不同窗口间数据的干扰。

我们在上面的代码中使用了后面要讲解的 ListState 。这里可以先简单说明一下。我们先声

明一个列表状态变量 :

private ListState<Event> UrlViewCountListState;

然后在 open 方法中初始化了列表状态变量，我们初始化的时候使用了 ListStateDescriptor

描述符，这个描述符用来告诉 Flink 列表状态变量的名字和类型。列表状态变量是单例，也就

是说只会被实例化一次。这个列表状态变量的作用域是当前 key 所对应的逻辑分区。我们使用

add 方法向列表状态变量中添加数据，使用 get 方法读取列表状态变量中的所有元素。

另外，根据水位线的定义，我们这里的延迟时间设为 0 事实上也是可以保证数据都到齐的。

感兴趣的读者可以自行修改代码进行测试。 208

7.5 侧输出流（ Side Output ）

处理函数还有另外一个特有功能，就是将自定义的数据放入“侧输出流”（ side output ）输

出。这个概念我们并不陌生，之前在讲到窗口处理迟到数据时，最后一招就是输出到侧输出流。

而这种处理方式的本质，其实就是处理函数的侧输出流功能。

我们之前讲到的绝大多数转换算子，输出的都是单一流，流里的数据类型只能有一种。而

侧输出流可以认为是“主流”上分叉出的“支流”，所以可以由一条流产生出多条流，而且这

些流中的数据类型还可以不一样。利用这个功能可以很容易地实现“分流”操作。

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

的 .output() 方法就可以了。

DataStream<Integer> stream = env.addSource(...);

SingleOutputStreamOperator<Long> longStream =

stream.process(new

ProcessFunction<Integer, Long>() {

@Override

public void processElement( Integer value, Context ctx, Collector<Integer>

out) throws Exception {

// 转换成 Long ，输出到主流中

out.collect(Long.valueOf(value));

// 转换成 String ，输出到侧输出流中

ctx.output(outputTag, "side-output: " + String.valueOf(value));

}

});

这里 output() 方法需要传入两个参数，第一个是一个“输出标签” OutputTag ，用来标识侧

输出流，一般会在外部统一声明；第二个就是要输出的数据。

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

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

如果想要获取这个侧输出流，可以基于处理之后的 DataStream 直接调用 .getSideOutput()

方法，传入对应的 OutputTag ，这个方式与窗口 API 中获取侧输出流是完全一样的。

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

7.6 本章总结

Flink 拥有非常丰富的多层 API ，而底层的处理函数可以说是最为强大、最为灵活的一种。

广义上来讲，处理函数也可以认为是 DataStream API 中的一部分，它的调用方式与其他转换

算子完全一致。处理函数可以访问时间、状态，定义定时操作，它可以直接触及流处理最为本

质的组成部分。所以处理函数不仅是我们处理复杂需求时兜底的“大招”，更是理解流处理本

质的重要一环。

在本章中，我们详细介绍了处理函数的功能和底层的结构，重点讲解了最为常用的

KeyedProcessFunction 和 ProcessWindowFunction ，并实现了电商应用中 Top N 的经典案例，另

外还介绍了侧输出流的用法。而关于合并两条流之后的处理函数，以及广播连接流

（ BroadcastConnectedStream ）的处理操作，调用方法和原理都非常类似，我们会在后续章节继

续展开。

8. 多流转换

无论是基本的简单转换和聚合，还是基于窗口的计算，我们都是针对一条流上的数据进行

处理的。而在实际应用中，可能需要将不同来源的数据连接合并在一起处理，也有可能需要将

一条流拆分开，所以经常会有对多条流进行处理的场景。本章我们就来讨论 Flink 中对多条流

进行转换的操作。

简单划分的话，多流转换可以分为“分流”和“合流”两大类。目前分流的操作一般是通

过侧输出流（ side output ）来实现，而合流的算子比较丰富，根据不同的需求可以调用 union 、

connect 、 join 以及 coGroup 等接口进行连接合并操作。下面我们就进行具体的讲解。

8.1 分流

所谓“分流”，就是将一条数据流拆分成完全独立的两条、甚至多条流。也就是基于一个

DataStream ，得到完全平等的多个子 DataStream ，如图 8-1 所示。一般来说，我们会定义一些

筛选条件，将符合条件的数据拣选出来放到对应的流里。

8.1.1 简单实现

其实根据条件筛选数据的需求，本身非常容易实现：只要针对同一条流多次独立调

用 .filter() 方法进行筛选，就可以得到拆分之后的流了。

例如，我们可以将电商网站收集到的用户行为数据进行一个拆分，根据类型（ type ）的不

同，分为“ Mary ”的浏览数据、“ Bob ”的浏览数据等等。那么代码就可以这样实现：

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class SplitStreamByFilter {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 SingleOutputStreamOperator<Event> stream = env
 .addSource(new ClickSource());
 // 筛选 Mary 的浏览行为放入 MaryStream 流中
 DataStream<Event> MaryStream = stream.filter(new FilterFunction<Event>() 
{
 @Override
 public boolean filter(Event value) throws Exception {
 return value.user.equals("Mary");
 }
 });
 // 筛选 Bob 的购买行为放入 BobStream 流中
 DataStream<Event> BobStream = stream.filter(new FilterFunction<Event>() {
210
 @Override
 public boolean filter(Event value) throws Exception {
 return value.user.equals("Bob");
 }
 });
 // 筛选其他人的浏览行为放入 elseStream 流中
 DataStream<Event> elseStream = stream.filter(new FilterFunction<Event>() 
{
 @Override
 public boolean filter(Event value) throws Exception {
 return !value.user.equals("Mary") && !value.user.equals("Bob") ;
 }
 });
 MaryStream.print("Mary pv");
 BobStream.print("Bob pv");
 elseStream.print("else pv");
 env.execute();
 }
 
}

输出结果是：

Bob pv> Event{user='Bob', url='./home', timestamp=2021-06-23 17:30:57.388}
else pv> Event{user='Alice', url='./home', timestamp=2021-06-23 17:30:58.399}
else pv> Event{user='Alice', url='./home', timestamp=2021-06-23 17:30:59.409}
Bob pv> Event{user='Bob', url='./home', timestamp=2021-06-23 17:31:00.424}
else pv> Event{user='Alice', url='./prod?id=1', timestamp=2021-06-23 
17:31:01.441}
else pv> Event{user='Alice', url='./prod?id=1', timestamp=2021-06-23 
17:31:02.449}
Mary pv> Event{user='Mary', url='./home', timestamp=2021-06-23 17:31:03.465}

这种实现非常简单，但代码显得有些冗余——我们的处理逻辑对拆分出的三条流其实是一

样的，却重复写了三次。而且这段代码背后的含义，是将原始数据流 stream 复制三份，然后

对每一份分别做筛选；这明显是不够高效的。我们自然想到，能不能不用复制流，直接用一个

算子就把它们都拆分开呢？

在早期的版本中， DataStream API 中提供了一个 .split() 方法，专门用来将一条流“切分”

成多个。它的基本思路其实就是按照给定的筛选条件，给数据分类“盖戳”；然后基于这条盖

戳之后的流，分别拣选想要的“戳”就可以得到拆分后的流。这样我们就不必再对流进行复制

了。不过这种方法有一个缺陷：因为只是“盖戳”拣选，所以无法对数据进行转换，分流后的

数据类型必须跟原始流保持一致。这就极大地限制了分流操作的应用场景。现在 split 方法已

经淘汰掉了，我们以后分流只使用下面要讲的侧输出流。

8.1.2 使用侧输出流

在 Flink 1.13 版本中，已经弃用了 .split() 方法，取而代之的是直接用处理函数（ process

function ）的侧输出流（ side output ）。

我们知道，处理函数本身可以认为是一个转换算子，它的输出类型是单一的，处理之后得

到的仍然是一个 DataStream ；而侧输出流则不受限制，可以任意自定义输出数据，它们就像从

“主流”上分叉出的“支流”。尽管看起来主流和支流有所区别，不过实际上它们都是某种类型

的 DataStream ，所以本质上还是平等的。利用侧输出流就可以很方便地实现分流操作，而且得

到的多条 DataStream 类型可以不同，这就给我们的应用带来了极大的便利。

关于处理函数中侧输出流的用法，我们已经在 7.5 节做了详细介绍。简单来说，只需要调

用上下文 ctx 的 .output() 方法，就可以输出任意类型的数据了。而侧输出流的标记和提取，都

离不开一个“输出标签”（ OutputTag ），它就相当于 split() 分流时的“戳”，指定了侧输出流的

id 和类型。

我们可以使用侧输出流将上一小节的分流代码改写如下：

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
public class SplitStreamByOutputTag {
 // 定义输出标签，侧输出流的数据类型为三元组(user, url, timestamp)
 private static OutputTag<Tuple3<String, String, Long>> MaryTag = new 
OutputTag<Tuple3<String, String, Long>>("Mary-pv"){};
 private static OutputTag<Tuple3<String, String, Long>> BobTag = new 
OutputTag<Tuple3<String, String, Long>>("Bob-pv"){};
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 SingleOutputStreamOperator<Event> stream = env
 .addSource(new ClickSource());
 SingleOutputStreamOperator<Event> processedStream = stream.process(new 
ProcessFunction<Event, Event>() {
 @Override
 public void processElement(Event value, Context ctx, Collector<Event> 
out) throws Exception {
 if (value.user.equals("Mary")){
 ctx.output(MaryTag, new Tuple3<>(value.user, value.url, 
value.timestamp));
 } else if (value.user.equals("Bob")){
212
 ctx.output(BobTag, new Tuple3<>(value.user, value.url, 
value.timestamp));
 } else {
 out.collect(value);
 }
 }
 });
 processedStream.getSideOutput(MaryTag).print("Mary pv");
 processedStream.getSideOutput(BobTag).print("Bob pv");
 processedStream.print("else");
 env.execute();
 }
}

输出结果是：

Bob pv> (Bob,./prod?id=1,1624442886645)
Mary pv> (Mary,./prod?id=1,1624442887664)
Bob pv> (Bob,./home,1624442888673)
Mary pv> (Mary,./prod?id=1,1624442889676)
else> Event{user='Alice', url='./prod?id=1', timestamp=2021-06-23 18:08:10.693}
else> Event{user='Alice', url='./prod?id=1', timestamp=2021-06-23 18:08:11.697}
else> Event{user='Alice', url='./prod?id=1', timestamp=2021-06-23 18:08:12.702}
Mary pv> (Mary,./cart,1624442893705)
Bob pv> (Bob,./cart,1624442894710)
else> Event{user='Alice', url='./cart', timestamp=2021-06-23 18:08:15.722}
Mary pv> (Mary,./prod?id=1,1624442896725)

这里我们定义了两个侧输出流，分别拣选 Mary 的浏览事件和 Bob 的浏览事件；由于类型

已经确定，我们可以只保留 ( 用户 id, url, 时间戳 ) 这样一个三元组。而剩余的事件则直接输出

到主流，类型依然保留 Event ，就相当于之前的 elseStream 。这样的实现方式显然更简洁，也

更加灵活。

8.2 基本合流操作

既然一条流可以分开，自然多条流就可以合并。在实际应用中，我们经常会遇到来源不同

的多条流，需要将它们的数据进行联合处理。所以 Flink 中合流的操作会更加普遍，对应的

API 也更加丰富。

8.2.1 联合（ Union ）

最简单的合流操作，就是直接将多条流合在一起，叫作流的“联合”（ union ），如图 8-2

所示。联合操作要求必须流中的数据类型必须相同，合并之后的新流会包括所有流中的元素，

数据类型不变。这种合流方式非常简单粗暴，就像公路上多个车道汇在一起一样。

在代码中，我们只要基于 DataStream 直接调用 .union() 方法，传入其他 DataStream 作为参

数，就可以实现流的联合了；得到的依然是一个 DataStream ：

stream1.union(stream2, stream3, ...)

注意： union() 的参数可以是多个 DataStream ，所以联合操作可以实现多条流的合并。

这里需要考虑一个问题。在事件时间语义下，水位线是时间的进度标志；不同的流中可能

水位线的进展快慢完全不同，如果它们合并在一起，水位线又该以哪个为准呢？

还以要考虑水位线的本质含义，是“之前的所有数据已经到齐了”；所以对于合流之后的

水位线，也是要以最小的那个为准，这样才可以保证所有流都不会再传来之前的数据。换句话

说，多流合并时处理的时效性是以最慢的那个流为准的。我们自然可以想到，这与之前介绍的

并行任务水位线传递的规则是完全一致的；多条流的合并，某种意义上也可以看作是多个并行

任务向同一个下游任务汇合的过程。

我们可以用下面的代码做一个简单测试：

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import java.time.Duration;
public class UnionExample {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
214
 
SingleOutputStreamOperator<Event> stream1 = env.socketTextStream("hadoop102", 
7777)
 .map(data -> {
 String[] field = data.split(",");
 return new Event(field[0].trim(), field[1].trim(), 
Long.valueOf(field[2].trim()));
 })
 .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBound
edOutOfOrderness(Duration.ofSeconds(2))
 .withTimestampAssigner(new 
SerializableTimestampAssigner<Event>() {
 @Override
 public long extractTimestamp(Event element, long 
recordTimestamp) {
 return element.timestamp;
 }
 })
 );
 stream1.print("stream1");
 SingleOutputStreamOperator<Event> stream2 = 
env.socketTextStream("hadoop103", 7777)
 .map(data -> {
 String[] field = data.split(",");
 return new Event(field[0].trim(), field[1].trim(), 
Long.valueOf(field[2].trim()));
 })
 .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBound
edOutOfOrderness(Duration.ofSeconds(5))
 .withTimestampAssigner(new 
SerializableTimestampAssigner<Event>() {
 @Override
 public long extractTimestamp(Event element, long 
recordTimestamp) {
 return element.timestamp;
 }
 })
 );
 stream2.print("stream2");
 // 合并两条流
 stream1.union(stream2)
 .process(new ProcessFunction<Event, String>() {
 @Override
 public void processElement(Event value, Context ctx, 
Collector<String> out) throws Exception {
 out.collect(" 水 位 线 ： " + 
ctx.timerService().currentWatermark());
 }
 })
 .print();
 env.execute();
 }
}

这里为了更清晰地看到水位线的进展，我们创建了两条流来读取 socket