水位线生成策略

水位线是事件时间处理的步调控制机制。在分布式流处理环境中，它们通过提供一个进度衡量标准来解决乱序数据问题。带有时间戳 $t$ 的水位线表明，直到时间 $t$ 的数据流已处理完毕。这一断言使窗口操作符能够关闭分区并触发计算，前提是不会再有时间戳 $t' \le t$ 的事件到达。

这种机制本质上是启发式的。由于网络分区和可变延迟使得完全了解数据流变得不可能，水位线策略体现了延迟和正确性之间的权衡。一种激进的策略，在接近当前处理时间生成水位线，会带来低延迟的结果，但有将有效数据作为“迟到数据”丢弃的风险。一种保守的策略会等待较长时间以处理滞留事件，确保更高的准确性，但会增加延迟。

水位线策略接口

在 Flink 中，WatermarkStrategy 定义了水位线的生成方式以及时间戳如何分配给事件。该接口将时间戳提取逻辑与水位线发送逻辑解耦。虽然简单的应用可能依赖预定义策略，但生产管线通常需要调整这些参数 (parameter)，以适应上游数据源（如 Kafka 分区）特定的数据倾斜情况。

水位线生成主要有两种方式：周期性生成和打点式生成。

周期性水位线

周期性生成是高吞吐量 (throughput)数据流的标准方法。系统会以由自动水位线间隔定义的常规处理时间间隔发送水位线。该间隔与事件到达速率无关。即使每秒有数千个事件到达，水位线也可能仅每 200 毫秒更新一次，以减少开销。

框架会按这些间隔调用 WatermarkGenerator 的 onPeriodicEmit 方法。生成器会检查目前已看到的最大时间戳，并根据滞后容忍度发送一个水位线。

打点式水位线

打点式生成依赖于数据本身。生成器会观察每个传入事件，并可能选择立即发送水位线。这在数据稀疏或特定标记 (token)记录表示一个逻辑块（例如混合流中的“批处理结束”信号）完成的场景中是必要的。这种方法会带来更高的开销，因为 onEvent 方法必须为每个记录评估水位线逻辑。

有界乱序

在像 Kafka 这样的分布式日志中最常见的模式是有界乱序。该策略假定元素虽然会乱序，但它们在有限的时间窗内发生。如果系统中的最大延迟已知为 5 秒，那么在任何时间点，水位线都是观察到的最大时间戳减去 5 秒。

$W = \max(T_{事件}) - T_{容忍}$

在此容忍缓冲期内到达的事件会被正确处理。水位线通过事件时间戳后到达的事件被认为是迟到事件。

下图说明了事件时间、处理时间与水位线之间的关系。水位线以下的点代表将被丢弃或发送到侧输出的迟到事件。

蓝色线代表水位线边界。绿色点在窗口内处理。红色点在水位线通过其时间戳后到达，被归类为迟到事件。

实现自定义策略

尽管 Flink 提供了 WatermarkStrategy.forBoundedOutOfOrderness，复杂的 AI 管线通常需要自定义逻辑。例如，你可能需要一个基于最后 $N$ 个时间戳平均值而不是严格最大值来推进时间以过滤异常值的策略，或者一个处理单个分区中交错的不同逻辑流的策略。

要实现自定义策略，你需要定义一个 WatermarkGenerator。

public class SmoothLagGenerator implements WatermarkGenerator<MyEvent> {
    
    private final long maxLag;
    private long currentMaxTimestamp;

    public SmoothLagGenerator(long maxLag) {
        this.maxLag = maxLag;
        this.currentMaxTimestamp = Long.MIN_VALUE + maxLag + 1;
    }

    @Override
    public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
        // 跟踪目前已看到的最大时间戳
        currentMaxTimestamp = Math.max(currentMaxTimestamp, eventTimestamp);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // 发送减去滞后的水位线
        output.emitWatermark(new Watermark(currentMaxTimestamp - maxLag));
    }
}

该生成器随后附加到流源：

WatermarkStrategy<MyEvent> strategy = WatermarkStrategy
    .forGenerator(ctx -> new SmoothLagGenerator(Duration.ofSeconds(5).toMillis()))
    .withTimestampAssigner((event, timestamp) -> event.getCreationTime());

DataStream<MyEvent> stream = env.fromSource(kafkaSource, strategy, "Kafka 源");

水位线传播与同步

在并行流处理拓扑中，水位线必须跨分区同步。流操作符，例如 Window ProcessFunction，通常从上游操作符的多个并行实例接收输入。每个上游实例根据其观察到的数据发送自己的水位线。

下游操作符为每个输入通道维护一个特定于分区的水位线。操作符的逻辑时间定义为所有输入通道水位线的最小值。这确保了操作符的时间不会超过“最慢”的上游分区。

$W_{操作符} = \min(W_{分区\_0}, W_{分区\_1}, \dots, W_{分区\_n})$

这种机制导致对最慢分区的依赖。如果某个 Kafka 分区因网络问题或数据倾斜而严重滞后，则该操作符的整个窗口操作将停止。

窗口操作符通过取所有传入通道水位线的最小值来确定其全局水位线。这里，分区 2 将全局进度滞留在 12:02。

处理空闲源

水位线传播中的一个值得关注的边界情况是空闲源问题。如果某个 Kafka 分区长时间没有数据，它将不会发送水位线。由于下游操作符计算所有分区的最小值，空闲分区缺少水位线会阻止最小值增加。整个应用的时间实际上会停滞，窗口无法关闭。

这通常发生在为低频类别保留的主题分区中，或在流量较低的时期。为解决此问题，Flink 提供了一个空闲检测机制。

WatermarkStrategy<MyEvent> strategy = WatermarkStrategy
    .forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withIdleness(Duration.ofMinutes(1));

当配置 withIdleness 时，如果在指定持续时间内没有事件到达，系统会将分区标记 (token)为“空闲”。一旦被标记为空闲，该分区将从下游水位线计算中排除。这使得全局水位线仅根据活跃分区前进。当数据从空闲分区恢复流动时，它会恢复活跃状态并再次参与同步逻辑。

仔细配置空闲超时时间是必要的。如果设置过短，系统可能会在空闲和活跃状态之间频繁切换。如果设置过长，当分区实际静默时，管线延迟会显著增加。在数据速率因分区而异的任何环境中，此配置是生产就绪的强制检查项。