企业官网建设流程全解析

最近帮团队升级一个实时数据任务时，编译环境直接给了一堆弃用警告，我点进去一看，AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks都被划了删除线。这意味着之前那套 Event Time + Watermark 的老写法，在新版 Flink 里已经正式走入了淘汰流程。如果你刚接触 Flink，或者还在沿用早期视频教程里的代码，这篇文章就是给你准备的：我会把新版本 Watermark 的使用方式、为什么要这么改、以及从老 API 迁移时最容易翻车的细节，一次性讲清楚。

1. 新版本 Watermark API 到底改了什么：旧接口的时代结束了

1.1 被废弃的老接口发生了什么

老版本的 Watermark 使用，核心是AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks这两个抽象类。它们做的事情看起来很简单：从数据里提取时间戳，然后通过回调方法返回一个 Watermark 对象。

assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Order>() { @Override public long extractTimestamp(Order element, long previousElementTimestamp) { return element.getEventTime(); } @Override public Watermark getCurrentWatermark() { return new Watermark(System.currentTimeMillis() - 5000); } });

这段代码放在 Flink 1.10 时代完全没毛病，但在新版里会直接提示已废弃。从 1.17 开始官方就标记了这两个接口并计划在后续版本移除，核心原因有两个：

第一，职责耦合。老接口把“从记录里提取时间戳”和“基于当前数据生成水印”两件事塞在同一个类里，表面上方便，实际用起来很容易搞混。很多人写出的getCurrentWatermark根本没利用已处理的数据，而是用系统时钟做减法，这在水印语义上是不对的。系统时钟是处理时间，不是事件时间的参考系。用System.currentTimeMillis()生成水印，一旦上游延迟或任务积压，水印会和真实事件进度严重脱节。

第二，扩展能力差。老接口无法支持 Watermark 对齐、空闲源处理这类高级策略。早期 Flink 版本里处理“某个分区没数据导致整个窗口卡住”的问题，只能靠手写自定义函数，既不优雅也容易出错。

1.2 新接口从底层换了一套设计思路

新版本把水印的逻辑统一收敛到了WatermarkStrategy里。它的设计核心是分层解耦：

WatermarkStrategy ├── TimestampAssigner (从数据中提取事件时间) └── WatermarkGenerator (根据已看到的所有事件时间推进水印)

WatermarkStrategy本身只是策略容器，它可以被配置给DataStream的assignTimestampsAndWatermarks方法，也可以直接配给新 Source API 的fromSource。你不再是继承某个类然后覆写方法，而是组合不同的策略组件。

底层逻辑变了，使用方式也必须跟着变。常见到的一个错误是：有人把WatermarkStrategy当成老Assigner的简单换皮，只换了入口，里面还是用系统时钟生成水印。这样即使代码编译通过，语义也是错的。新版本的价值不在于换了个 API 名字，而在于强迫你重新思考“水印到底代表什么”——它代表的是“在此之前的数据都已经到达了”的推断，不是墙上挂钟的时间。

2. WatermarkStrategy 的正确打开姿势：先分清四个角色再动手

2.1 TimestampAssigner：谁来提取事件时间

TimestampAssigner负责从数据中提取 Event Time。这是水印的起点，水印好不好用，首先要看事件时间提取得准不准。

WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((order, timestamp) -> order.getEventTime());

这里order.getEventTime()就是订单数据的业务时间。有个细节：提取出来的时间戳是毫秒。如果你上游字段是秒，记得乘以 1000。我在实际项目里遇到过一次数据源给的是秒级时间戳，结果整个窗口迟了 1000 倍才触发，排查了几个小时才发现在这个不起眼的地方。

withTimestampAssigner也可以单独实现成一个类，比如从联合主键中拼接时间、从嵌套 POJO 中取字段、甚至调用外部服务补齐时间，但一般不推荐在时间戳分配器里做重逻辑，它会被每条数据调用，太重会影响吞吐。

2.2 WatermarkGenerator：谁来推进水印

WatermarkGenerator负责维护水印状态。官方内置了两个开箱即用的策略：

• forMonotonousTimestamps()：假设事件时间单调递增，适合近乎有序的数据流。
• forBoundedOutOfOrderness(Duration)：容忍乱序，允许最大乱序范围内的迟到数据。

它们的本质区别在于：单调递增策略每看到一条记录就把水印推进到该记录的事件时间；乱序容忍策略则是把水印推进到“已见最大事件时间 - 乱序容忍度”。后者更通用，因为真实业务数据几乎都有乱序。

自定义生成器时，需要实现onEvent和onPeriodicEmit两个方法。onEvent在每条数据到达时调用，用来观察事件时间并更新内部状态；onPeriodicEmit则周期性发射当前水印。注意：不要把水印推进逻辑都塞到onPeriodicEmit里而忽略onEvent，因为周期性发射时如果没有保存“已见最大时间戳”的状态，水印就永远推不上去。

2.3 周期性发射 vs 点式发射：选错的影响

新版里WatermarkGenerator通过WatermarkOutput发射水印，发射方式分为两种：

周期长度可以通过env.getConfig().setAutoWatermarkInterval(Duration.ofMillis(200))调整，默认是 200ms（不同小版本有差异，建议显式设置）。周期越短，水印越接近实时数据，但也会增加发射频率和下游触发次数，属于吞吐与延迟的 trade-off。

官方默认的WatermarkStrategy采用周期性发射，内部通过WatermarkGenerator接口统一处理。如果你在onEvent里直接调用output.emitWatermark，就相当于手工实现了点式逻辑，不推荐，除非你有非常明确的“看到某类型数据必须立刻推水印”的业务需求。

2.4 Idleness 与 Watermark 对齐：新版里的关键配置

新版本水印使用中，最容易忽略但影响最大的是两个配置：空闲源处理和多源水印对齐。

空闲源处理对应withIdleness(Duration)。场景是：一个 Kafka topic 有多个分区，其中某个分区长时间没有新数据。由于 Flink 下游水印取所有输入分区的“最小值”，空转分区的水印不推进，整个窗口就永远不触发。withIdleness会让超过指定时间没有新数据的输入被标记为空闲，从而从水印合并中剔除。

WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withIdleness(Duration.ofMinutes(1)) .withTimestampAssigner(...)

这个配置强烈建议加上，尤其是当你使用 Kafka 多分区或机房双链路冗余数据源的时候。我见过因为没配withIdleness，数据源切换演练后整个分钟级窗口延迟了一个多小时才触发的现场，加了之后立刻恢复正常。

多源水印对齐对应withWatermarkAlignment(String groupId, Duration maxDrift, Duration updateInterval)。它解决的是多个时间戳来源（比如不同主题、不同数据源）之间水印进度差异过大的问题，要求这些数据源在“相同的事件时间语义”下对齐推进，避免快的等慢的导致长尾延迟。这个在生产链路里很有用，但一般单作业用不到，属于有多源合并需求时的进阶配置。

3. 可直接抄的代码：新版本水印应用的三个实战场景

3.1 基础场景：带乱序容忍的 Kafka 订单流

最常规的写法是把WatermarkStrategy直接应用到从 Kafka 读取的 DataStream 上，然后做事件时间滚动窗口。

import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.connector.kafka.source.KafkaSource; import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; import java.time.Duration; StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.getConfig().setAutoWatermarkInterval(Duration.ofMillis(200)); KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("broker-1:9092,broker-2:9092") .setTopics("order-topic") .setGroupId("order-app-group") .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStream<String> raw = env.fromSource( source, WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((event, ts) -> Long.parseLong(parseOrderTime(event))) .withIdleness(Duration.ofMinutes(1)), "kafka-order-source" );

env.fromSource方法的第二个参数就是 WatermarkStrategy，这是新版 Source API 的推荐入口。它比老式的addSource更合理：水印策略在 Source 层面就被绑定，Source 的分片读取器会基于各自的数据进度生成水印，再由框架统一合并。

这里需要解释一下为什么forBoundedOutOfOrderness(Duration.ofSeconds(10))配 10 秒。这个值不是拍脑袋的，它代表你乐意等待乱序数据到达的时间窗口。如果业务数据从日志产生到进入 Kafka 的链路耗时高峰是 6 秒，那你容忍度配 10 秒会覆盖大多数情况；配小了窗口提前触发，迟到数据被丢弃，配大了窗口延迟触发，下游报表数据晚出。我需要确认的是：你的“乱序容忍度”应该略大于数据源 p99 延迟差，而不是无限放大。

3.2 多源水印对齐的高级场景

假如你有两个数据源：一个支付流水 topic，一个退款流水 topic，窗口计算需要把两者按事件时间关联。单独为每个源配置水印后，下游会取两个源的当前水印的最小值。如果支付源的数据实时性远高于退款源，退款源稍有抖动，支付源的水印就会被拖住，整体延迟飙升。

WatermarkStrategy<Payment> paymentStrategy = WatermarkStrategy .<Payment>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withWatermarkAlignment("trade-group", Duration.ofSeconds(30), Duration.ofSeconds(5)) .withTimestampAssigner((payment, ts) -> payment.getPayTime()); WatermarkStrategy<Refund> refundStrategy = WatermarkStrategy .<Refund>forBoundedOutOfOrderness(Duration.ofSeconds(30)) .withWatermarkAlignment("trade-group", Duration.ofSeconds(30), Duration.ofSeconds(5)) .withTimestampAssigner((refund, ts) -> refund.getRefundTime());

withWatermarkAlignment的语义是：属于同一个对齐组的多个水印，如果进度差异超过maxDrift，就强制快的等待慢的，避免无限拉开差距。updateInterval表示检查对齐状态的频率。

这个功能不是让你无脑加的，它适合两个流必须保持“事件时间同步推进”的场景。如果两个源本身对时间语义要求不一致，硬套对齐反而会拖垮整体吞吐。实测建议：先不加对齐跑几天，看监控里两个源的水印差距是否超过你的业务容忍线，再决定是否开启。

3.3 窗口触发后：迟到数据的兜底处理

水印推进后，不等于所有数据都准时到达。即使有乱序容忍度，仍然会有极端迟到数据落在窗口触发之后。新版本里处理迟到数据的推荐做法是sideOutputLateData，把迟到数据单独接入一条旁路流，而不是直接丢弃。

OutputTag<Order> lateTag = new OutputTag<Order>("late-orders") {}; DataStream<Order> windowed = orders .keyBy(order -> order.getOrderId()) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .sideOutputLateData(lateTag) .apply(new WindowFunction<Order, Order, Long, TimeWindow>() { @Override public void apply(Long key, TimeWindow window, Iterable<Order> input, Collector<Order> out) { for (Order order : input) { out.collect(order); } } }); windowed.getSideOutput(lateTag).map(...).print();

这里有个容易混淆的细节：旁路流只是给你一个“处理迟到数据”的出口，它不会重新触发窗口计算。如果你需要对迟到数据重新计算窗口，就得用allowedLateness配合后期触发，那是另一套更复杂的语义。常见做法是：窗口主链路做准实时输出，迟到旁路流做分钟级的修正任务，后台再对账一次。这样既保证了实时性，也不丢数据。Flink 新版本中，这属于“主链路快速处理 + 旁路精准修正”的组合拳，比一味调大乱序容忍要健康得多。

4. 从旧 API 迁移时最容易翻车的几个细节

4.1 迁移前后的代码长什么样

老版本典型的周期水印写法：

DataStream<Order> orders = env .addSource(kafkaConsumer) .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Order>() { private long maxSeen = Long.MIN_VALUE; @Override public long extractTimestamp(Order element, long previousTimestamp) { maxSeen = Math.max(maxSeen, element.getEventTime()); return element.getEventTime(); } @Override public Watermark getCurrentWatermark() { return new Watermark(maxSeen - 5000); } });

迁移到新版：

DataStream<Order> orders = env .fromSource(kafkaSource, WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((order, ts) -> order.getEventTime()), "order-source");

如果你只是把AssignerWithPeriodicWatermarks平移到WatermarkStrategy.forBoundedOutOfOrderness，大部分场景可以无痛切换，因为后者内部实现的正是“已见最大时间戳 - 容忍度”的周期发射逻辑。

但真正的坑不在 API 表面，而在迁移后的行为差异上。

4.2 迁移后水印不推进：一次完整的排查过程

之前在某公司做数据平台重构时，把一套老订单任务从 Flink 1.12 升到新版本，迁移完跑起来后窗口迟迟不触发。我当时的排查链路如下，你可以直接拿来参考。

第一步，看 Web UI 里水印是否推进。如果所有 subtask 的水印都是Long.MIN_VALUE，说明水印生成器根本没收到数据，或者时间戳提取出来是负值。我那次排查的结果是：新版本SimpleStringSchema反序列化后是字符串，时间戳字段是本地时间字符串格式，而不是毫秒数。也就是说时间戳分配器解析失败了，事件时间没读出来。老代码里 extractTimestamp 也是同样逻辑，但当时的测试数据刚好带了一个可解析字段，掩盖了问题。

第二步，看数据是否确实进入了算子。通过map算子打日志确认 Source 下游有数据流，排除 Kafka 消费组配置错误、offset 重置等问题。

第三步，检查是否有某个并行子任务长时间无数据。我当时发现问题在withIdleness没加：任务有 24 个并行度，其中两个 Kafka 分区因为主题 key 分布不均几乎没数据，导致整体水印一直停留在初始状态。加上withIdleness(Duration.ofMinutes(1))后，水印立刻开始正常推进。

4.3 并行度认识错误导致的水印卡死

另一个常见问题是把“全局水印”误理解为“每个算子独立的水印”。Flink 的水印合并逻辑是：每个并行子任务都有自己的当前水印，上游多个并行流进入下游算子时，取的是所有输入通道水印的最小值。

常见翻车操作是：在keyBy之后的窗口前调用类似并行度调整或重分区操作，导致某个并行通道一直没数据。比如某任务在keyBy前有 32 并行度，keyBy后只设置了 4 并行度，部分 Kafka 分区数据稀疏，某些下游通道长时间没有摄入数据。由于水印是逐通道转发的，空转通道会卡住该通道上的窗口。这时候withIdleness也能兜底，但更根本的解法是让并行度与数据分布匹配，或者使用rescale做局部重平衡，而不是靠大水印容忍度硬扛。

4.4 与老版本行为差异最隐蔽的一处：事件时间的丢弃语义

老版本中，如果一条记录的时间戳早于当前水印，它会在进入窗口算子的输入缓存时被标记为迟到数据。新版本里，这个过程被提前到了 Source 阶段：WatermarkStrategy在 Source 端生成水印，Source 读取器会丢弃那些明显迟于当前水印的早到数据。这意味着：迁移后，某些 Edge 场景下你能在下游算子看到的迟到数据变少了，因为更早的数据在源头就被过滤。这其实是更合理的优化，但对老业务来说，突然少了的“迟到修正数据”可能影响原有的修正链路，需要留意监控里 late 数据量的变化趋势。

5. 新版本 Watermark 使用中的经验与建议

5.1 日常操作中必须养成的三个检查习惯

第一，开任务前先确认时间语义。env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)在较新版本中已经被移除，不需要也不能再设置。有同事在升级时找不到这个入口慌了，以为新版本不支持事件时间了，其实是语义内置了，EventTime是默认且唯一的时间语义，不需要显式声明。但ProcessingTime仍然可以通过TimeWindow的参数使用，别混淆。

第二，每个水印任务都建议在开发环境打印一次水印推进情况。可以通过ProcessFunction的context.timerService().currentWatermark()输出当前水印，或者直接用 Web UI 查看“Watermark”列。我的习惯是在核心窗口前放一个map算子，每 10 秒打印当前水印和最大事件时间，这样任务刚启动时就能快速判断时间戳分配逻辑是否正常。

第三，注意检查autoWatermarkInterval是否被调整过。如果任务里有人改成了一个很大的值，比如 10 秒，水印虽然最终会推进，但窗口触发时间可能比实际数据落后接近 10 秒，这在准实时业务里很容易被忽略，因为性能指标看起来一切正常。

5.2 关于乱序容忍度、窗口大小和业务容忍度的“三角平衡”

我个人的推荐流程是：先收集一批真实数据，统计从数据产生到进入 Flink 之间的事件时间延迟分布，分别看 p50、p95、p99。然后把乱序容忍度设到大于 p99 延迟差的值，而不是习惯性地填 5 秒或 10 秒。窗口大小则根据业务口径定，比如分钟级实时大屏用TumblingEventTimeWindows.of(Time.minutes(1))，不要用大窗口加大容忍度来兜底迟到，那样会让数据实时性大打折扣。

实际踩过一次坑：某指标需要 5 分钟窗口，业务要求数据延迟不超过 1 分钟，我当时把乱序容忍度配成了 5 分钟，等于窗口要等 5 分钟才能触发，业务完全不满意。后来改成容忍度 30 秒，窗口还是 5 分钟，辅以旁路迟到流做修正，延迟立刻降下来了。容忍度不是越大越好，它只是事件时间语义下对不确定性的一种折中。

5.3 最后再分享一个实用小技巧

如果你用的数据源是自己实现的，而不是 Kafka，在新版本里依然可以用老经典的方式addSource，但要记得把WatermarkStrategy当成Source的属性来配置，而不是在DataStream上去额外调用assignTimestampsAndWatermarks。你可以在自定义 Source 的run方法里直接调用SourceContext.emitWatermark，对于有确定时间步进的测试数据源来说，这比走onEvent+onPeriodicEmit那套更直接。

不过生产环境我基本只用两种 Watermark 策略：forBoundedOutOfOrderness处理一般乱序流，自定义生成器处理需要特殊业务规则的流。其他花哨的写法，除非有明确的度量数据和业务理由，否则不建议引入，因为水印策略是全局生效的，一个不合理的手写生成器会把整个数据链路的时间语义带偏。新版本的 Watermark 设计已经足够合理，你要做的不是发明新用法，而是理解它背后的推进模型，然后用正确的方式把事件时间表达出来。