Flink流处理：如何聚合JOIN与EXPLODE生成的全量记录？

针对你遇到的问题——处理Kafka实时数据时，JOIN+EXPLODE生成多条记录后需要筛选某字段最大值的记录，且避免会话窗口带来的延迟，这里提供两种更优的实现方式：

方案一：使用Flink SQL的窗口函数（推荐） #

利用Flink SQL的ROW_NUMBER()窗口函数，结合水印机制确保同一unique_id的所有衍生记录都到达后再输出结果，完全避免窗口延迟。

实现步骤： #

1. 为流表定义水印（基于事件时间），让Flink能识别同一unique_id的所有记录是否已到齐：

CREATE TABLE joined_msip (
    unique_id STRING,
    target_field INT, -- 要取最大值的字段
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '0.1' SECOND -- 根据实际数据延迟调整
) WITH (
    'connector' = 'kafka', -- 你的源连接器配置
    'topic' = 'xxx',
    'properties.bootstrap.servers' = 'xxx',
    'format' = 'json',
    ...
);

2. 用窗口函数筛选每个unique_id下target_field最大的记录：

SELECT *
FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (PARTITION BY unique_id ORDER BY target_field DESC) AS rn
    FROM joined_msip
) t
WHERE rn = 1;

3. 将结果写入目标存储，配置触发模式确保全量记录到达后输出：

CREATE TABLE sink_table (
    unique_id STRING,
    target_field INT,
    ... -- 其他需要的字段
) WITH (
    'connector' = 'xxx', -- 目标存储连接器（如kafka、jdbc等）
    'sink.emit.mode' = 'after-watermark', -- 水印触发后输出，避免提前输出不完整结果
    ...
);

INSERT INTO sink_table
SELECT unique_id, target_field, ...
FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (PARTITION BY unique_id ORDER BY target_field DESC) AS rn
    FROM joined_msip
) t
WHERE rn = 1;

如果同一unique_id的衍生记录是瞬时生成的，可将水印延迟设为极小值，几乎不会产生额外延迟。

方案二：使用DataStream API的KeyedProcessFunction #

通过自定义KeyedProcessFunction，为每个unique_id维护状态收集所有衍生记录，当确认记录全量到达后计算最大值并输出。

实现示例： #

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class MaxValueProcessFunction extends KeyedProcessFunction<Object, Row, Row> {
    private transient ListState<Row> recordState;

    @Override
    public void open(FunctionInitializationContext context) throws Exception {
        // 初始化状态，存储同一unique_id的所有衍生记录
        ListStateDescriptor<Row> descriptor = new ListStateDescriptor<>("recordState", Row.class);
        recordState = context.getOperatorStateStore().getListState(descriptor);
    }

    @Override
    public void processElement(Row value, Context ctx, Collector<Row> out) throws Exception {
        recordState.add(value);
        // 注册极短定时器，确保同一unique_id的所有记录都已进入状态
        ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 10);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Row> out) throws Exception {
        // 遍历状态找到target_field最大的记录
        Row maxRow = null;
        int maxValue = Integer.MIN_VALUE;
        for (Row row : recordState.get()) {
            int currentValue = (int) row.getField("target_field");
            if (currentValue > maxValue) {
                maxValue = currentValue;
                maxRow = row;
            }
        }
        // 输出结果并清空状态
        if (maxRow != null) {
            out.collect(maxRow);
        }
        recordState.clear();
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        // 状态快照，保证容错
        recordState.get();
    }
}

调用该函数： #

DataStream<Row> joined = tableEnv.toDataStream(tableEnv.from("joined_msip"));

DataStream<Row> filtered = joined
        .keyBy((KeySelector<Row, Object>) value -> value.getField("unique_id"))
        .process(new MaxValueProcessFunction());

这种方式通过极短定时器触发计算，相比会话窗口的延迟几乎可以忽略，同时保证能收集到同一unique_id的所有衍生记录。

为什么会话窗口方案有延迟？ #

会话窗口需要等待会话间隙时间内没有新记录到来才会触发聚合，这就导致必须等待设定的间隙时间（比如你用的1000ms）才能输出结果，而上面的两种方案要么通过水印确认记录全量到达，要么用极短的定时器触发，大幅降低了延迟。

内容的提问来源于stack exchange，提问作者AnastaDol