Flink大尺寸滚动窗口AggregateFunction被异常拆分问题求助

问题根因 #

• 核心触发原因是Flink历史版本的已知缺陷：Flink 1.12及更早版本中，TumblingEventTimeWindows计算窗口边界时使用32位整型存储时间偏移量，当窗口长度超过*24.85天（对应Integer.MAX_VALUE毫秒值2147483647ms）*时，会发生整数溢出，导致窗口边界计算错误，本该连续的单个窗口被拆分为多个不规则小窗口触发计算。
• 你测试的1小时、7天窗口对应毫秒值分别为3600000ms、604800000ms，均远小于24.85天的阈值，因此运行正常；30天对应2592000000ms、31天对应2678400000ms，均超过阈值触发溢出bug，和你观测到的拆分现象完全吻合。该问题和内存、状态后端配置无关，因此调整状态后端、监控JVM堆内存均无法解决。
• 额外逻辑偏差：即使修复溢出问题，Time.days(30)/Time.days(31)实现的是固定天数滚动窗口，并非自然月窗口，不同月份天数差异会导致窗口和自然月不对齐。
• 现有代码缺陷：AvgQ1的merge方法仅合并了count和sum字段，没有处理last_timestamp，发生窗口合并场景时，会导致最后取到的时间戳不是窗口内最新事件的时间戳，影响时间槽计算结果。

排查步骤 #

• 核对Flink集群运行版本，版本低于1.13时优先匹配上述整数溢出bug。
• 在窗口聚合前追加ProcessWindowFunction打印当前窗口的start/end时间戳，直接确认窗口是否被错误切分，排除下游逻辑误判的可能。
• 检查水位线上报逻辑，确认没有因为数据源断流、乱序等待配置过大导致水位线异常推进提前触发窗口（你观测到的拆分区间长度加总正好等于30/31天，基本可以排除水位线问题）。
• 检查Kafka消费位点配置，确认作业启动时没有从中间位点开始消费，导致部分数据落在窗口触发点之后触发拆分，从最早位点重启作业即可复现/排除该问题。

解决方案 #

1. 版本升级修复内核bug
   升级Flink集群到1.13及以上版本，该版本后窗口边界计算全部改用64位长整型存储时间值，彻底解决长窗口整数溢出问题。
2. 自然月窗口实现
   如果需要按自然月维度聚合，不要使用固定天数的滚动窗口，推荐两种实现方式：

• 自定义WindowAssigner，直接根据事件时间戳计算所属自然月的起止时间分配窗口，逻辑精准可控。
• 使用KeyedProcessFunction注册自然月结束时间的定时器，在定时器触发时输出当月聚合结果，灵活度更高，方便处理迟到数据。

3. 修复聚合函数逻辑
   补全merge方法的字段合并逻辑，合并时取两个累加器中更大的last_timestamp作为合并后的时间戳，避免时间字段异常：

@Override
public AccumulatorQ1 merge(AccumulatorQ1 a, AccumulatorQ1 b) {
    a.count += b.count;
    a.sum += b.sum;
    // 保留两个累加器中最新的事件时间戳
    a.last_timestamp = Math.max(a.last_timestamp, b.last_timestamp);
    return a;
}

4. 临时规避方案（无法升级版本时使用）
   如果暂时不能升级Flink版本，可以将月窗口拆分为多个连续的7天窗口做预聚合，下游再按自然月维度做二次聚合，绕过单窗口长度超过24.85天的溢出阈值。

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