Flink Watermark 设计分析

Flink Watermark 演进分析

1. 核心痛点：如何衡量事件时间进度？

在乱序流中，直接使用“当前看到的最大时间戳”作为进度会导致窗口过早关闭。系统需要一种机制来声明：“我认为这个时间点之前的数据已经全部到齐”。

Watermark 就是这个保守下界声明。

2. 演进一：提取与生成分离（Watermark 诞生）

• 痛点：真实数据源通常包含两类时间信息：事件自带的时间戳、数据源愿意推进的时间进度。
• 机制：将这两者解耦为TimestampAssigner和WatermarkGenerator。
• 源码映射：统一入口为WatermarkStrategy，由TimestampsAndWatermarksOperator在运行时驱动：
  • 每条数据调用onEvent()更新观察值。
  • 周期性调用onPeriodicEmit()发射 Watermark。
• 注意事项：
  • Watermark 必须单调递增，一旦对下游生效，TimestampsAndWatermarksOperator会直接丢弃更小的值。

3. 演进二：多输入的木桶效应（取最小值）

• 痛点：多分区或多输入时，如果快慢分区进度不一致，全局进度不能以最快的分区为准，否则会导致慢分区数据被误判为迟到。
• 机制：取所有活跃输入中的最小值作为全局 Watermark。
• 源码映射：StatusWatermarkValve维护一个对齐分片的最小堆，只有当前堆顶（最小值）变大时，才会向下游发射新的 Watermark (StatusWatermarkValve.java#L273-L285)。

4. 演进三：部分分区断流拖死全局（Idle 机制）

• 痛点：如果某个分区并非处理慢，而是迟迟没有新数据（断流），它的 Watermark 会停滞，导致全局 Watermark 卡死，下游窗口无法触发。
• 机制：通过withIdleness()标记空闲，以牺牲该分区数据的正确性风险来换取系统整体的活性（Liveness）。
• 源码映射：StatusWatermarkValve在收到IDLE状态后，将该分片从最小堆中剔除。
• 代价与恢复（诈尸处理）：
  • 移除分区意味着全局 Watermark 可能会比真实情况偏大。如果该分区随后突然苏醒并发送旧数据，由于全局 Watermark 已无法回退，这些数据将被无情地判定为迟到数据（Late Data）。
  • 恢复进堆条件：当分区苏醒（收到ACTIVE信号）时，并不会立刻重新参与对齐计算。只有当该分区新的 Watermark大于等于当前的全局 Watermark（即追上大部队）时，才允许重新加入最小堆 (StatusWatermarkValve.java#L250-L261)。对于因此产生的迟到数据，需结合演进五的兜底机制（Allowed Lateness / Side Output）来处理。

5. 演进四：快慢分区导致状态膨胀（Watermark Alignment）

• 痛点：Idle 机制只能解决“完全物理断流”的死锁，但无法解决“龟速活跃分区”的问题。如果 B 分区并未断流（一直在发旧数据），只是其时间戳进度极其陈旧（比如停在12:00），它永远不会触发 Idle 超时。此时，若 A 分区极快地读到了13:00，全局 Watermark 依然会被活跃的 B 分区死死卡在12:00。A 分区超前读取的海量数据将无法触发清理，只能一直积压在下游状态中，最终导致 OOM。注意：造成积压的根本原因不是“绝对速度太快”，而是分区之间的“相对进度差异太大”。
• 疑问：为什么不能主动把龟速分区标为 IDLE 踢掉？：API 上确实可以主动markIdle()，但这会引发灾难性的语义崩塌。因为 B 分区是活跃的，如果把它踢掉，全局 Watermark 会瞬间跟着 A 跃升到13:00并触发窗口清理。随后 B 正常吐出的12:00到13:00之间的海量合法数据，将全部被误判为**迟到数据（Late Data）**被丢弃。主动 IDLE 是用牺牲海量数据的正确性来换取不 OOM，这在业务上通常是不可接受的。
• 机制：既然不能抛弃慢分区的数据（保正确性），又不能让快分区继续堆状态（保稳定性），唯一的解法就是给快分区“踩刹车”（Watermark Alignment）。
• 源码映射：通过withWatermarkAlignment()设置允许的最大偏差（maxDrift）。当某个源的相对超前量超过阈值（即myWatermark > groupMin + maxDrift）时，框架会强行暂停该快分区的底层读取（Pause Reading），以此牺牲系统的整体吞吐量来换取状态大小的安全可控。

6. 演进五：Watermark 误判与迟到数据处理（Allowed Lateness / Side Output）

• 痛点：Watermark 只是启发式边界，如果真实数据比 Watermark 还要晚到（真实迟到），直接丢弃会导致计算结果不准确。
• 机制 1：Allowed Lateness（延长状态寿命）
  • 原理：允许窗口在触发后，不立即清理状态，而是保留到window.end + allowedLateness。在这期间到来的迟到数据，会再次触发窗口计算并更新结果。
  • 源码映射：WindowOperator的cleanupTime()返回window.maxTimestamp() + allowedLateness。只有 Watermark 超过这个清理时间时，才会注册清理 Timer (WindowOperator.java#L671-L677)。
• 机制 2：Side Output（兜底侧输出）
  • 原理：对于超过allowedLateness彻底迟到的数据，不再污染主数据流，而是打入侧输出流（Side Output），交由离线或兜底逻辑处理。
  • 源码映射：在WindowOperator中，如果isElementLate(element)为 true，则直接调用sideOutput(element)发送到lateDataOutputTag(WindowOperator.java#L587-L589)。

7. 常见误区与干预边界

• Watermark 单调性不可回退：无法通过手动发一个极小的 Watermark 来“召回”迟到数据，因为框架层会直接忽略。挽救迟到数据只能靠Allowed Lateness。
• 触发器逻辑分离：Watermark 本身不执行业务逻辑，它只负责更新 Timer Service 的currentWatermark，进而触发所有<= currentWatermark的 Event-Time Timer，最终由 Timer 驱动窗口触发。
• 避免过度手动发 Watermark：除非是自定义 Source 且外部系统带有明确的进度信号（如 Binlog 封口），否则业务算子不应主动发射 Watermark，应交由WatermarkStrategy在 Source 端统一生成。