一、为什么需要水位线？

事件在真实世界里可能乱序到达。Flink 以**水位线（Watermark）**来“宣告”事件时间的推进：当下游收到 Watermark(T)，就可认为将不再收到时间戳 < T 的事件，从而触发窗口计算、定时器等。

约定：所有时间戳与水位线均是自 1970-01-01T00:00:00Z 起的毫秒数（epoch millis）。

二、WatermarkStrategy：一体化策略的核心

WatermarkStrategy 同时包含：

• TimestampAssigner：为事件提取/分配事件时间；
• WatermarkGenerator：生成水位线。

常用策略通过静态方法直接可用，也可以把自定义的时间戳分配器与生成器“打包”成一个策略。

示例：有界乱序 + Lambda 时间戳分配器

WatermarkStrategy
    .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
    .withTimestampAssigner((event, ts) -> event.f0);

贴士：多数情况下无需显式指定 TimestampAssigner，如 Kafka/Kinesis 可直接使用记录自身的时间戳。

三、策略放在哪：Source 上 vs. 算子后

首选在 Source 上配置（更精确）：

• 能利用分片/分区（split/partition）粒度的知识，产出更准确的整体水位线；
• 需使用 Source 特定接口（例如 Kafka FLIP-27 Source）。

无法在 Source 设置时，可在算子后设置：

DataStream<MyEvent> withWm = stream
    .filter(e -> e.severity() == WARNING)
    .assignTimestampsAndWatermarks(<watermark strategy>);

注意：若原流已有时间戳/水位线，这里会覆盖它们。

四、空闲分区（Idle Source）与状态膨胀

当某些分区长期无数据时，其水位线不前进，会拖住整体水位线（整体是所有并行水位线的最小值）。
解决：启用空闲检测，把该输入标记为 idle，避免拖慢整体进度。

WatermarkStrategy
    .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
    .withIdleness(Duration.ofMinutes(1));

五、水位线对齐（Watermark Alignment）

相反，若某些分区特别快，会导致下游（如窗口/Join）缓存过多快分区数据，状态可能失控。
解决：开启对齐，限制某个来源的水位线超前幅度（“最大漂移”）。

WatermarkStrategy
    .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
    .withWatermarkAlignment("alignment-group-1",
                            Duration.ofSeconds(20),  // 允许的最大漂移
                            Duration.ofSeconds(1));  // 汇报频率

• 仅 FLIP-27 Source 支持；在 Source 之后用 assignTimestampsAndWatermarks 不生效。
• 同组的 Source 共享同一“对齐分组”标签；Flink 会暂停超前的分区，等待落后方推进。
• 1.17 起支持split 级对齐，连接器需实现暂停/恢复 split。如需兼容旧版本/不支持 split 级对齐，可通过
  pipeline.watermark-alignment.allow-unaligned-source-splits=true 关闭 split 级对齐（有行为限制，见官方说明）。

六、自定义 WatermarkGenerator：两种风格

WatermarkGenerator 有两种典型风格：周期式（periodic）与插桩式（punctuated）。

6.1 周期式（Periodic）

按 ExecutionConfig.setAutoWatermarkInterval(...) 的周期触发 onPeriodicEmit 发水位线；onEvent 用于记录观测值。

有界乱序示例

public class BoundedOutOfOrdernessGenerator implements WatermarkGenerator<MyEvent> {
    private final long maxOutOfOrderness = 3500; // 3.5s
    private long currentMaxTs;

    @Override
    public void onEvent(MyEvent e, long ts, WatermarkOutput out) {
        currentMaxTs = Math.max(currentMaxTs, ts);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput out) {
        out.emitWatermark(new Watermark(currentMaxTs - maxOutOfOrderness - 1));
    }
}

固定滞后处理时间示例

public class TimeLagWatermarkGenerator implements WatermarkGenerator<MyEvent> {
    private final long maxTimeLag = 5000; // 5s

    @Override public void onEvent(MyEvent e, long ts, WatermarkOutput out) { /* no-op */ }

    @Override public void onPeriodicEmit(WatermarkOutput out) {
        out.emitWatermark(new Watermark(System.currentTimeMillis() - maxTimeLag));
    }
}

6.2 插桩式（Punctuated）

遇到带标记的事件即时发水位线；onPeriodicEmit 通常不需实现。

public class PunctuatedAssigner implements WatermarkGenerator<MyEvent> {
    @Override
    public void onEvent(MyEvent e, long ts, WatermarkOutput out) {
        if (e.hasWatermarkMarker()) {
            out.emitWatermark(new Watermark(e.getWatermarkTimestamp()));
        }
    }
    @Override public void onPeriodicEmit(WatermarkOutput out) { /* no-op */ }
}

性能提示：不要每个事件都发水位线；水位线会触发下游计算，过多会显著拖慢流水线。

七、Kafka 分区感知水位线（强烈推荐）

多分区并行消费会打乱单分区的时间序列模式。Flink 的 Kafka 分区感知策略在 **KafkaSource 内“每分区”**生成水位线，再按与 shuffle 相同的合并规则合并，能最大化保留单分区的“严格递增/有界乱序”特性。

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers(brokers)
    .setTopics("my-topic")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> stream = env.fromSource(
    source,
    WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(20)),
    "mySource");

这里未指定 TimestampAssigner，直接使用 Kafka 记录的时间戳。

八、算子如何处理水位线（窗口触发的顺序规则）

• 单输入算子：必须先处理完当前水位线所触发的所有输出，再向下游发送水位线本身。
• 双输入算子：算子的当前水位线取两输入水位线的最小值；处理顺序同上。
• 具体行为由 processWatermark 系列方法定义（OneInputStreamOperator#processWatermark、TwoInputStreamOperator#processWatermark1/2）。

九、迁移与兼容性：旧版 Assigner 接口

历史上 Flink 使用 AssignerWithPeriodicWatermarks 与 AssignerWithPunctuatedWatermarks。
建议全部迁移至 WatermarkStrategy + TimestampAssigner + WatermarkGenerator：关注点更清晰，统一两种风格，API 也更易组合与维护。

十、工程实践清单

1. 策略优先级：尽量在 Source 上设置 WatermarkStrategy；无法设置时再放在算子后。
2. 有界乱序估计：从上游系统观测 95/99 分位的延迟分布，给 forBoundedOutOfOrderness 留安全余量。
3. 空闲检测：多分区/多文件源务必 withIdleness(...)，避免“木桶效应”。
4. 对齐分组：同一“业务流”的多 Source（如 Kafka + File）用同一对齐组标签，限制最大漂移；密切观察 RPC 增加带来的开销。
5. 生成频率：合理设置 setAutoWatermarkInterval；太频繁会影响吞吐，太稀疏会拉高延迟。
6. 自定义生成器：周期式优先；插桩式仅在上游确有“标记事件”时使用。不要为每条事件发水位线。
7. 窗口正确性：牢记“水位线先触发数据输出，后传递自身”的顺序，调试窗口边界问题时尤为关键。
8. Kafka 最佳实践：优先使用 FLIP-27 KafkaSource + 分区感知水位线；确认并行度与分区数的映射关系，避免分区分配不均带来的水位线异象。
9. 观测与报警：接入水位线进度、对齐暂停次数、状态大小、背压等指标；异常早发现早处理。
10. 版本/兼容：升级 1.17 及以上关注 split 级对齐支持；若连接器暂不支持，使用配置开关降级行为。

十一、结语

把握好策略位置（Source 优先）、乱序容忍度估计、空闲与对齐控制以及算子顺序规则，你的事件时间作业既能在乱序中保持正确性，也能在高吞吐下稳定运行。下一步，建议结合你的业务流量与延迟分布，做一版真实数据回放来校准 out-of-orderness 与 watermark interval——这往往是把“理论正确”变成“生产可用”的关键一步。