Flink双流联结与处理函数详解

一、前言

在上一篇文章中，我们深入学习了Flink的时间语义与水位线机制。本文将继续探索Flink流处理的高级特性——双流联结与处理函数。

双流联结解决了"如何将两条数据流按照某种条件关联起来"的问题；处理函数则提供了Flink最底层的编程接口，可以访问时间、状态和侧输出流，实现任意复杂的业务逻辑。

二、基于时间的双流联结（Join）

2.1 为什么需要双流Join

在实际业务中，我们经常需要将两条流的数据按照某个条件关联起来。例如：

• 将订单流与支付流关联，统计支付成功率
• 将用户点击流与商品曝光流关联，计算转化率
• 将传感器数据流与告警规则流关联，实时告警

Flink提供了两种基于时间的双流联结方式：窗口联结（Window Join） 和 间隔联结（Interval Join）。

2.2 窗口联结（Window Join）

窗口联结将两条流的数据分配到相同的时间窗口中，然后在窗口内对满足关联条件的数据进行配对处理。

上图展示了窗口联结的基本原理：两条流的数据被分配到同一个时间窗口内，只有落在同一窗口且满足关联条件的数据才会被匹配输出。

2.2.1 窗口联结的调用方式

stream1.join(stream2)
.where(<KeySelector>) // 第一条流的key
.equalTo(<KeySelector>) // 第二条流的key
.window(<WindowAssigner>) // 窗口分配器
.apply(<JoinFunction>); // 联结函数

核心要点：

• where() 和 equalTo() 分别指定两条流的关联key
• .window() 指定窗口类型（滚动、滑动、会话）
• .apply() 传入 JoinFunction，处理匹配的数据对
• 窗口Join本质上是等值内联结（Inner Join），只有匹配上的数据才会输出

2.2.2 窗口联结实例

public class WindowJoinDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

// 第一条流：用户点击事件 (user, timestamp)
SingleOutputStreamOperator<Tuple2<String, Long>> ds1 = env
.fromElements(
Tuple2.of("a", 1000L),
Tuple2.of("a", 2000L),
Tuple2.of("b", 3000L),
Tuple2.of("c", 4000L)
)
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple2<String, Long>>forMonotonousTimestamps()
.withTimestampAssigner((value, ts) -> value.f1)
);

// 第二条流：订单事件 (user, orderId, timestamp)
SingleOutputStreamOperator<Tuple3<String, String, Long>> ds2 = env
.fromElements(
Tuple3.of("a", "order1", 1000L),
Tuple3.of("a", "order2", 1500L),
Tuple3.of("b", "order3", 3000L),
Tuple3.of("d", "order4", 5000L)
)
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple3<String, String, Long>>forMonotonousTimestamps()
.withTimestampAssigner((value, ts) -> value.f2)
);

// Window Join
DataStream<String> join = ds1.join(ds2)
.where(r1 -> r1.f0) // ds1的key：user
.equalTo(r2 -> r2.f0) // ds2的key：user
.window(TumblingEventTimeWindows.of(Time.seconds(5))) // 5秒滚动窗口
.apply(new JoinFunction<Tuple2<String, Long>,
Tuple3<String, String, Long>, String>() {
@Override
public String join(Tuple2<String, Long> first,
Tuple3<String, String, Long> second) throws Exception {
return first + " <——> " + second;
}
});

join.print();
env.execute();
}
}

窗口Join的特点：

• 落在同一个时间窗口范围内才能匹配
• 根据 keyBy 的 key 进行匹配关联
• 只能拿到匹配上的数据，类似固定时间范围的 inner join
• 未匹配的数据不会输出

2.3 间隔联结（Interval Join）

窗口联结要求数据落在同一个固定窗口内，但在某些场景下，匹配的时间间隔可能不是固定的。例如：订单发生后10分钟内找到对应的支付记录。

间隔联结针对一条流的每个数据，开辟出其时间戳前后的一段时间间隔，看这期间是否有来自另一条流的数据匹配。

2.3.1 间隔联结的原理

给定两个时间点：下界（lowerBound） 和 上界（upperBound）。对于流A中的数据元素a，可以匹配的时间区间为：

[a.timestamp + lowerBound, a.timestamp + upperBound]

如果流B中的数据元素b的时间戳落在该区间内，则a和b匹配成功。

上图展示了间隔联结的时间窗口概念：对于 Orders 流中的每个事件，定义一个时间区间，只有落在该区间的 Shipments 事件才能匹配。

匹配条件：

a.timestamp + lowerBound <= b.timestamp <= a.timestamp + upperBound

2.3.2 间隔联结的调用方式

stream1
.keyBy(<KeySelector>)
.intervalJoin(stream2.keyBy(<KeySelector>))
.between(Time.milliseconds(–2), Time.milliseconds(1))
.process(new ProcessJoinFunction<>() {...});

核心要点：

• 必须基于 KeyedStream 调用
• between(lowerBound, upperBound) 指定时间间隔
• 目前只支持事件时间
• 两条流关联后的 watermark 以两条流中最小的为准

2.3.3 间隔联结实例

public class IntervalJoinDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

// 第一条流：下单事件 (userId, timestamp)
SingleOutputStreamOperator<Tuple2<String, Long>> orderStream = env
.fromElements(
Tuple2.of("user1", 1000L),
Tuple2.of("user1", 5000L),
Tuple2.of("user2", 2000L),
Tuple2.of("user3", 6000L)
)
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple2<String, Long>>forMonotonousTimestamps()
.withTimestampAssigner((value, ts) -> value.f1)
);

// 第二条流：支付事件 (userId, amount, timestamp)
SingleOutputStreamOperator<Tuple3<String, Double, Long>> payStream = env
.fromElements(
Tuple3.of("user1", 99.9, 1500L),
Tuple3.of("user1", 199.9, 5500L),
Tuple3.of("user2", 59.9, 2500L),
Tuple3.of("user4", 299.9, 8000L)
)
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple3<String, Double, Long>>forMonotonousTimestamps()
.withTimestampAssigner((value, ts) -> value.f2)
);

// Interval Join：订单发生后 [-2秒, +2秒] 内的支付记录
orderStream
.keyBy(r -> r.f0)
.intervalJoin(payStream.keyBy(r -> r.f0))
.between(Time.seconds(–2), Time.seconds(2))
.process(new ProcessJoinFunction<
Tuple2<String, Long>,
Tuple3<String, Double, Long>,
String>() {
@Override
public void processElement(
Tuple2<String, Long> left, // orderStream的数据
Tuple3<String, Double, Long> right, // payStream的数据
Context ctx,
Collector<String> out) throws Exception {
out.collect("订单：" + left + " 匹配到支付：" + right);
}
})
.print();

env.execute();
}
}

2.3.4 处理迟到数据

间隔联结也支持将迟到数据输出到侧输出流：

OutputTag<Tuple2<String, Long>> orderLateTag =
new OutputTag<>("order-late", Types.TUPLE(Types.STRING, Types.LONG)){};
OutputTag<Tuple3<String, Double, Long>> payLateTag =
new OutputTag<>("pay-late", Types.TUPLE(Types.STRING, Types.DOUBLE, Types.LONG)){};

SingleOutputStreamOperator<String> process = orderStream
.keyBy(r -> r.f0)
.intervalJoin(payStream.keyBy(r -> r.f0))
.between(Time.seconds(–2), Time.seconds(2))
.sideOutputLeftLateData(orderLateTag) // 左流迟到数据
.sideOutputRightLateData(payLateTag) // 右流迟到数据
.process(new ProcessJoinFunction<>() {...});

process.print("主流");
process.getSideOutput(orderLateTag).printToErr("订单迟到数据");
process.getSideOutput(payLateTag).printToErr("支付迟到数据");

Interval Join 的注意事项：

• 只支持事件时间
• 指定上界、下界的偏移，负号代表时间往前，正号代表时间往后
• process中只能处理匹配上的数据
• 两条流关联后的 watermark 以两条流中最小的为准

三、处理函数（ProcessFunction）

3.1 处理函数概述

之前学习的转换算子（map/filter/flatMap）和窗口操作，都是基于 DataStream 的转换。在 Flink 更底层，我们可以不定义任何具体算子，而是提炼出一个统一的"处理"操作——这就是处理函数（ProcessFunction）。

上图展示了 Flink API 的层次结构：ProcessFunction 位于最底层，提供了最灵活的处理能力。DataStream Conversion Operations 图则展示了各种流类型之间的转换关系。

处理函数的核心能力：

• 访问事件时间戳和水位线信息
• 注册定时器（Timer），实现基于时间的回调
• 访问状态（State）
• 将数据输出到侧输出流（Side Output）

3.2 ProcessFunction

ProcessFunction 是最基本的处理函数，基于 DataStream 直接调用 .process() 方法使用。

3.2.1 接口定义

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 {}
}

3.2.2 核心方法解析

processElement() 方法：

• 对于流中的每个元素都会调用一次
• value：当前输入元素
• ctx：上下文对象，包含时间戳、TimerService、侧输出流等
• out：收集器，用于输出结果

onTimer() 方法：

• 注册好的定时器触发时调用
• timestamp：定时器触发的时间戳
• 事件时间语义下由水位线触发
• 处理时间语义下由系统时间触发

3.2.3 ProcessFunction 实例

public class ProcessFunctionDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

env.fromElements(1, 2, 3, 4, 5)
.process(new ProcessFunction<Integer, String>() {
@Override
public void processElement(Integer value, Context ctx,
Collector<String> out) throws Exception {
// 获取当前数据的事件时间戳
long timestamp = ctx.timestamp();
// 获取当前处理时间
long processingTime = ctx.timerService().currentProcessingTime();
// 获取当前水位线
long watermark = ctx.timerService().currentWatermark();

out.collect("数据：" + value +
"，事件时间：" + timestamp +
"，处理时间：" + processingTime +
"，水位线：" + watermark);
}
})
.print();

env.execute();
}
}

3.3 KeyedProcessFunction

上图展示了 KeyedProcessFunction 在 Flink 算子层次中的位置，以及 Function 与 State 的关系。处理函数可以访问和修改状态，实现复杂的有状态计算。

KeyedProcessFunction 是对 KeyedStream 调用的处理函数，相比 ProcessFunction 增加了对 TimerService 的支持，可以注册和删除定时器。

3.3.1 TimerService 详解

上图展示了 TimerService 的工作机制：通过 TimerService 注册事件时间或处理时间定时器，当时间进展到设定值时，定时器触发，调用 onTimer() 方法。

TimerService 提供的方法：

// 获取当前处理时间
long currentProcessingTime();

// 获取当前的水位线（事件时间）
long currentWatermark();

// 注册处理时间定时器
void registerProcessingTimeTimer(long time);

// 注册事件时间定时器
void registerEventTimeTimer(long time);

// 删除处理时间定时器
void deleteProcessingTimeTimer(long time);

// 删除事件时间定时器
void deleteEventTimeTimer(long time);

定时器去重机制：
TimerService 会以 key + 时间戳 为标准对定时器进行去重。对于每个 key 和时间戳，最多只有一个定时器。

3.3.2 KeyedProcessFunction 实例

public class KeyedProcessTimerDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction())
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((element, ts) -> element.getTs() * 1000L)
);

KeyedStream<WaterSensor, String> sensorKS =
sensorDS.keyBy(sensor -> sensor.getId());

SingleOutputStreamOperator<String> process = sensorKS.process(
new KeyedProcessFunction<String, WaterSensor, String>() {

@Override
public void processElement(WaterSensor value, Context ctx,
Collector<String> out) throws Exception {
// 获取当前数据的key
String currentKey = ctx.getCurrentKey();

// 获取TimerService
TimerService timerService = ctx.timerService();

// 事件时间案例：注册一个5秒后的定时器
Long currentEventTime = ctx.timestamp();
timerService.registerEventTimeTimer(currentEventTime + 5000L);
System.out.println("当前key=" + currentKey +
"，当前时间=" + currentEventTime +
"，注册了一个5s后的定时器");

// 处理时间案例：注册一个5秒后的定时器
// long currentTs = timerService.currentProcessingTime();
// timerService.registerProcessingTimeTimer(currentTs + 5000L);
}

@Override
public void onTimer(long timestamp, OnTimerContext ctx,
Collector<String> out) throws Exception {
super.onTimer(timestamp, ctx, out);
String currentKey = ctx.getCurrentKey();
System.out.println("key=" + currentKey +
"，现在时间是" + timestamp + "，定时器触发！");
}
}
);

process.print();
env.execute();
}
}

3.4 处理函数的分类

Flink 提供了8种不同的处理函数，适用于不同的场景：

四、实战案例：Top N

4.1 需求描述

实时统计一段时间内的出现次数最多的水位。例如，统计最近10秒钟内出现次数最多的两个水位，并且每5秒钟更新一次。

4.2 思路一：ProcessAllWindowFunction（不推荐）

不做 keyBy，直接基于 DataStream 开窗，使用全窗口函数统一处理。

public class ProcessAllWindowTopNDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction())
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((element, ts) -> element.getTs() * 1000L)
);

// 最近10秒 = 窗口长度，每5秒输出 = 滑动步长
sensorDS.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
.process(new MyTopNPAWF())
.print();

env.execute();
}

public static class MyTopNPAWF extends
ProcessAllWindowFunction<WaterSensor, String, TimeWindow> {

@Override
public void process(Context context, Iterable<WaterSensor> elements,
Collector<String> out) throws Exception {
// 使用HashMap统计各个vc出现的次数
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);
}
}

// 对count值进行排序
List<Tuple2<Integer, Integer>> datas = new ArrayList<>();
for (Integer vc : vcCountMap.keySet()) {
datas.add(Tuple2.of(vc, vcCountMap.get(vc)));
}

datas.sort((o1, o2) -> o2.f1 – o1.f1); // 降序排序

// 取出count最大的2个vc
StringBuilder outStr = new StringBuilder();
outStr.append("================================\\n");
for (int i = 0; i < Math.min(2, datas.size()); i++) {
Tuple2<Integer, Integer> vcCount = datas.get(i);
outStr.append("Top" + (i + 1) + "\\n");
outStr.append("vc=" + vcCount.f0 + "\\n");
outStr.append("count=" + vcCount.f1 + "\\n");
outStr.append("窗口结束时间=" +
DateFormatUtils.format(context.window().getEnd(),
"yyyy-MM-dd HH:mm:ss.SSS") + "\\n");
outStr.append("================================\\n");
}

out.collect(outStr.toString());
}
}
}

缺点：

• 并行度被强行设置为1，无法并行处理
• 需要收集齐所有数据后再遍历统计，效率低

4.3 思路二：KeyedProcessFunction（推荐）

先对数据做 keyBy 分区，分别统计每个 vc 的出现次数（增量聚合），然后再排序取 TopN。

public class KeyedProcessFunctionTopNDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction())
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((element, ts) -> element.getTs() * 1000L)
);

// 思路二：使用 KeyedProcessFunction 实现
// 1. 按照 vc 分组、开窗、聚合（增量计算 + 全量打标签）
SingleOutputStreamOperator<Tuple3<Integer, Integer, Long>> windowAgg =
sensorDS.keyBy(sensor -> sensor.getVc())
.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
.aggregate(
new VcCountAgg(), // 增量聚合：统计每个vc的出现次数
new WindowResult() // 全窗口函数：带上窗口结束时间标签
);

// 2. 按照窗口标签（窗口结束时间）keyBy，排序取 TopN
windowAgg.keyBy(r -> r.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 value, Integer accumulator) {
return accumulator + 1;
}

@Override
public Integer getResult(Integer accumulator) { return accumulator; }

@Override
public Integer merge(Integer a, Integer b) { return a + b; }
}

// 全窗口函数：给聚合结果打上窗口结束时间标签
public static class WindowResult extends
ProcessWindowFunction<Integer, Tuple3<Integer, Integer, Long>,
Integer, TimeWindow> {
@Override
public void process(Integer key, Context context, Iterable<Integer> elements,
Collector<Tuple3<Integer, Integer, Long>> out) throws Exception {
Integer count = elements.iterator().next();
long windowEnd = context.window().getEnd();
out.collect(Tuple3.of(key, count, windowEnd));
}
}

// KeyedProcessFunction：排序取 TopN
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;
private int threshold; // Top数量

public TopN(int threshold) {
this.threshold = threshold;
this.dataListMap = new HashMap<>();
}

@Override
public void processElement(Tuple3<Integer, Integer, Long> value,
Context ctx, Collector<String> out) throws Exception {
Long windowEnd = value.f2;

// 存到HashMap中
if (dataListMap.containsKey(windowEnd)) {
dataListMap.get(windowEnd).add(value);
} else {
List<Tuple3<Integer, Integer, Long>> dataList = new ArrayList<>();
dataList.add(value);
dataListMap.put(windowEnd, dataList);
}

// 注册定时器：windowEnd + 1ms
// 同一个窗口范围应该同时输出，延迟1ms即可
ctx.timerService().registerEventTimeTimer(windowEnd + 1);
}

@Override
public void onTimer(long timestamp, OnTimerContext ctx,
Collector<String> out) throws Exception {
Long windowEnd = ctx.getCurrentKey();

// 排序
List<Tuple3<Integer, Integer, Long>> dataList = dataListMap.get(windowEnd);
dataList.sort((o1, o2) -> o2.f1 – o1.f1); // 按count降序

// 取TopN
StringBuilder outStr = new StringBuilder();
outStr.append("================================\\n");
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");
}

// 清理用完的数据，节省资源
dataList.clear();
out.collect(outStr.toString());
}
}
}

优化思路：

• 先分区聚合再排序：先对每个 vc 做 keyBy + 增量聚合，并行统计各 vc 出现次数
• 再全局排序：按窗口结束时间 keyBy，使用 KeyedProcessFunction 收集所有 vc 的统计结果，排序取 TopN
• 定时器触发：注册 windowEnd + 1ms 的定时器，确保所有数据到齐后再排序输出

五、侧输出流（Side Output）

处理函数的另一个重要特性是可以将数据输出到侧输出流（Side Output）。

5.1 侧输出流的作用

• 实现数据分流：将符合条件的数据输出到侧输出流
• 处理异常数据：将格式错误的数据输出到侧输出流
• 收集迟到数据：窗口关闭后的迟到数据输出到侧输出流

5.2 侧输出流使用示例

public class SideOutputDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction())
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((element, ts) -> element.getTs() * 1000L)
);

// 定义侧输出流标签
OutputTag<String> warnTag = new OutputTag<>("warn", Types.STRING);

SingleOutputStreamOperator<WaterSensor> process = sensorDS
.keyBy(sensor -> sensor.getId())
.process(new KeyedProcessFunction<String, WaterSensor, WaterSensor>() {
@Override
public void processElement(WaterSensor value, 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("告警");

env.execute();
}
}

六、双流联结与处理函数对比总结

如果本文对你有帮助，欢迎 点赞 👍 + 收藏 ⭐ + 关注 🔖，你的支持是我持续创作的动力！