Flink窗口聚合异常：stream2的5分钟窗口包含stream1的前一分钟窗口

你遇到的问题核心在于ProcessingTime窗口的划分逻辑：

• ProcessingTime窗口完全基于算子处理数据的系统时间，而非数据本身对应的业务时间。
• stream1中00:59:00-00:59:59的窗口数据，可能因为实际处理延迟（比如上游数据到达晚、计算耗时等），导致这些数据到达stream2算子的系统时间落在了00:00:00之后，因此被划分到了stream2的00:00-00:04窗口中，而非预期的00:55-00:59窗口。

方案1：改用EventTime窗口（推荐） #

这是从根源解决问题的方式，让窗口基于数据的业务时间而非处理时间划分：

1. 给stream1的输出数据携带其原始窗口的时间戳（比如窗口的结束时间，作为该聚合结果的业务时间）。
2. 在stream2中使用EventTime窗口，并指定时间戳分配器，基于stream1输出的业务时间来划分窗口。

示例代码：

// 给stream1的聚合结果打上窗口结束时间的时间戳
DataStream<T> stream1WithTimestamp = stream1
    .assignTimestampsAndWatermarks(WatermarkStrategy.<T>forMonotonousTimestamps()
        .withTimestampAssigner((element, recordTimestamp) -> {
            // 假设element中包含其所属stream1窗口的结束时间，转换为毫秒时间戳
            return element.getWindowEndTime().toEpochMilli();
        }));

// stream2使用EventTime的滚动窗口
DataStream<T> stream2 = stream1WithTimestamp
    .keyBy(t -> t.id)
    .window(TumblingEventTimeWindows.of(Time.seconds(300)))
    .aggregate(new SomeAggregate());

方案2：ProcessingTime下的兼容处理 #

如果必须使用ProcessingTime，可以通过过滤或自定义窗口分配器来修正：

• 过滤法：在stream2的聚合前，先过滤掉业务时间不在当前窗口范围内的数据：

DataStream<T> stream2 = stream1
    .keyBy(t -> t.id)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(300)))
    .filter(element -> {
        // 校验element的业务时间是否落在当前窗口的时间范围内
        long windowStart = ...; // 获取当前窗口的开始时间（可通过上下文或预计算）
        long windowEnd = windowStart + 300000;
        return element.getWindowStartTime() >= windowStart && element.getWindowEndTime() < windowEnd;
    })
    .aggregate(new SomeAggregate());

• 自定义窗口分配器：基于stream1输出数据的业务时间，手动分配到对应的stream2窗口，替代默认的ProcessingTime窗口分配逻辑。

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