上一篇讲了 Flink 周期性 Watermark 的原理——按固定时间间隔默认 200ms触发 onPeriodicEmit 生成 Watermark是 Flink 默认且最常用的生成方式。这篇讲另一种生成方式断点式 WatermarkPunctuated Watermark。断点式 Watermark 不像周期性那样按固定时间间隔触发而是在每个事件到达时调用 onEvent根据事件内容判断是否满足触发条件如果满足则立即发出 Watermark。它的触发精度更高事件级但实现更复杂且如果没有触发事件Watermark 永远不推进。很多同学对断点式 Watermark 比较陌生因为生产环境 99% 的场景用周期性就够了。但在某些特殊场景事务边界事件、心跳事件、批次结束标记断点式 Watermark 能做到事件级精确触发显著减少窗口延迟。这篇从本质定义、触发机制、内部实现、四步生成流程、接口源码、四大适用场景、完整代码实现到常见问题和最佳实践把断点式 Watermark 一次性讲透。一、断点式 Watermark 本质onEvent 中根据事件内容触发断点式 WatermarkPunctuated Watermark是 Flink 中另一种 Watermark 生成方式。它在每个事件到达时调用onEvent(event, timestamp, output)方法根据事件内容判断是否满足触发条件如果满足则立即调用output.emitWatermark(new Watermark(timestamp))发出 Watermark。下面这张图把断点式 Watermark 的本质定义、触发时间线、与周期性的对比放在一起展示。1.1 核心特点第一Watermark 由特殊标记事件触发而不是固定时间间隔。断点式 Watermark 的触发完全取决于事件内容——只有特殊标记事件如事务边界事件、心跳事件、批次结束标记到达时才发出 Watermark普通事件只更新内部状态不发出 Watermark。第二触发精度高事件级。断点式 Watermark 在标记事件到达时立即发出不需要等待固定时间间隔所以触发精度是事件级的窗口触发延迟最小。第三实现复杂。需要在 onEvent 中判断事件内容区分普通事件和标记事件处理标记事件的时间戳保证 Watermark 单调递增。比周期性生成只需要更新 maxTs 周期性发出复杂得多。第四如果没有触发事件Watermark 永远不推进。这是断点式最大的风险。如果数据流中没有标记事件或标记事件丢失onEvent 中永远不会调用 emitWatermarkWatermark 永远不推进窗口永远不触发。1.2 适用场景断点式 Watermark 适用于数据流中有明确的时间标记事件的场景事务边界事件如数据库 binlog 的事务边界、支付系统的交易完成事件心跳/保活事件如监控系统的心跳包、IoT 设备的保活事件批次结束标记如文件结束标记、批量数据的结束事件、ETL 任务的完成标记分区/分片结束标记如 Kafka 分区的结束事件、Shard 的结束标记这些场景下标记事件明确表示这个时间之前的数据都到齐了遇到标记事件时立即发出 Watermark可以做到事件级精确触发减少窗口延迟。二、断点式 Watermark 的触发时间线理解断点式 Watermark 的关键是理解它的触发时间线时间轴 → 普通事件E1(10:00:00.000) → 普通事件E2(10:00:00.050) → 标记事件M1(10:00:00.100) → 发出WM(10:00:00.100) → 普通事件E3(10:00:00.200) → 普通事件E4(10:00:00.300) → 标记事件M2(10:00:00.400) → 发出WM(10:00:00.400)关键观察普通事件E1/E2/E3/E4到达时onEvent 只更新内部状态maxTimestamp不发出 Watermark标记事件M1/M2到达时onEvent 判断满足触发条件立即调用 emitWatermark 发出 WatermarkWatermark 发出后立即向下游传播可能触发窗口计算整个过程没有固定时间间隔完全由事件内容驱动2.1 与周期性的核心区别对比维度断点式 Punctuated周期性 Periodic触发时机每个事件到达时onEvent固定时间间隔默认200ms调用方法onEventonPeriodicEmit无数据时不会生成WM永远不推进仍会定期尝试生成但不推进实现复杂度较复杂需判断事件内容简单更新maxTs周期性发出推进精度高事件级精确一般间隔内不更新性能开销高每个事件都判断低间隔触发频率可控适用场景有明确时间标记事件大多数场景风险无触发事件时WM不推进无最稳定生产环境建议99% 的场景用周期性 Watermark有界乱序就够了稳定可靠。断点式只在有明确标记事件的特殊场景使用且建议配合周期性作为兜底避免标记事件丢失时 Watermark 不推进。三、断点式 Watermark 内部实现原理断点式 Watermark 涉及三个核心组件下面这张图把核心组件架构、四步生成流程、WatermarkGenerator 接口源码放在一起展示。3.1 核心组件一TimestampsAndWatermarksOperatorTimestampsAndWatermarksOperator是时间戳分配与 Watermark 生成的算子是assignTimestampsAndWatermarks()方法在底层创建的算子。它的职责接收上游数据调用TimestampAssigner从数据中提取事件时间戳每个事件到达时立即调用WatermarkGenerator.onEvent()断点式的核心不需要等待定时器将 Watermark 向下游传播注意与周期性不同断点式下 TimestampsAndWatermarksOperator 不需要注册 ProcessingTimeService 定时器因为 Watermark 完全由 onEvent 触发。3.2 核心组件二WatermarkGenerator.onEvent()WatermarkGeneratorT是 Watermark 生成器的核心接口。断点式生成方式下onEvent()是核心方法每个事件到达时立即调用 onEvent在 onEvent 中更新内部状态如 maxTimestamp判断当前事件是否是特殊标记事件如果是标记事件满足触发条件立即调用output.emitWatermark()发出 Watermark普通事件只更新状态不发出 Watermark3.3 核心组件三WatermarkOutput.emitWatermark()WatermarkOutput是 Watermark 的输出接口主要方法emitWatermark(Watermark watermark)发出一个 WatermarkmarkIdle()标记当前输入为空闲markActive()标记当前输入为活跃断点式生成方式下onEvent 中满足触发条件时调用output.emitWatermark(new Watermark(timestamp))发出 Watermark。发出前 Flink 内部会做单调递增检查确保 Watermark 只向前推进。四、四步生成流程详解从事件到达到 Watermark 发出完整的流程分为四步4.1 第一步事件到达 → 提取时间戳 → 调用 onEvent数据到达 TimestampsAndWatermarksOperator首先调用TimestampAssigner.extractTimestamp(event, previousTimestamp)从数据中提取事件时间戳然后调用WatermarkGenerator.onEvent(event, timestamp, output)。注意断点式生成方式下每个事件到达都会立即调用 onEvent不需要等待定时器。这是断点式与周期性的核心区别——周期性是等定时器触发 onPeriodicEmit断点式是事件到达立即触发 onEvent。4.2 第二步onEvent 判断事件内容 → 普通事件只更新状态在 onEvent 中首先更新内部状态如记录当前最大事件时间 maxTimestamp。然后判断当前事件是否是特殊标记事件。判断方式通常有事件类型字段如eventType WATERMARK_MARKER事件类如event instanceof WatermarkMarkerEvent特殊字段值如event.isWatermarkMarker() true普通事件只更新状态不发出 Watermark。只有特殊标记事件才会进入下一步。4.3 第三步满足触发条件 → 调用 emitWatermark 发出 Watermark如果当前事件是特殊标记事件满足触发条件立即调用output.emitWatermark(new Watermark(timestamp))发出 Watermark。Watermark 的时间戳通常是标记事件中携带的时间戳如事务提交时间或当前最大事件时间 maxTimestamp更安全保证单调递增发出是立即的不需要等待定时器这是断点式与周期性的核心区别。标记事件到达的瞬间Watermark 就发出了所以触发精度是事件级的。4.4 第四步单调递增检查 → Watermark 向下游传播 → 驱动窗口触发发出 Watermark 前Flink 内部会检查新 Watermark 是否大于当前 Watermark。Watermark 必须单调递增如果新 Watermark ≤ 当前 WatermarkFlink 会忽略这次发出。通过检查的 Watermark 向下游算子传播下游算子如 Window收到 Watermark 后检查是否有窗口的结束时间 ≤ Watermark如果有则触发这些窗口计算。五、WatermarkGenerator 接口源码解析// Flink 1.11 WatermarkGenerator 核心接口publicinterfaceWatermarkGeneratorT{// 每个事件到达时调用断点式生成的核心方法// 在这个方法中判断事件内容满足条件则调用 output.emitWatermark()voidonEvent(Tevent,longeventTimestamp,WatermarkOutputoutput);// 周期性调用默认200ms断点式生成方式下通常为空实现// 因为断点式完全由 onEvent 触发不需要周期性生成voidonPeriodicEmit(WatermarkOutputoutput);}断点式 WatermarkGenerator 的简化实现publicclassPunctuatedGeneratorimplementsWatermarkGeneratorMyEvent{privatelongmaxTimestampLong.MIN_VALUE;OverridepublicvoidonEvent(MyEventevent,longeventTimestamp,WatermarkOutputoutput){maxTimestampMath.max(maxTimestamp,eventTimestamp);// 判断是否是特殊标记事件如事务边界、心跳、批次结束if(event.isWatermarkMarker()){// 满足条件立即发出 Watermarkoutput.emitWatermark(newWatermark(maxTimestamp));}// 普通事件只更新 maxTimestamp不发出 Watermark}OverridepublicvoidonPeriodicEmit(WatermarkOutputoutput){// 断点式生成空实现完全由 onEvent 触发}}可以看到断点式实现的核心是onEvent 中更新 maxTimestamp判断事件是否是标记事件event.isWatermarkMarker()是标记事件则立即发出 Watermark用 maxTimestamp 保证单调递增onPeriodicEmit 空实现完全由 onEvent 触发六、断点式 Watermark 四大适用场景下面这张图把四大适用场景、六个常见问题、八条最佳实践放在一起展示。6.1 场景一事务边界事件数据流中有明确的事务提交/结束事件如数据库 binlog 的事务边界、支付系统的交易完成事件。遇到事务边界事件时立即发出 Watermark表示这个事务之前的数据都到齐了可以触发窗口计算。优势事务级精确触发窗口延迟最小结果准确。示例Canal 解析 MySQL binlog 时每个事务结束时有 XID 事件遇到 XID 事件发出 Watermark。6.2 场景二心跳/保活事件数据流中有定期发送的心跳事件如监控系统的心跳包、IoT 设备的保活事件。心跳事件中携带当前时间戳遇到心跳事件时发出 Watermark表示心跳时间之前的数据都到齐了。优势即使没有业务数据心跳事件也能推进 Watermark避免无数据时 Watermark 不推进。示例IoT 设备每 10 秒发送一次心跳包心跳包中携带设备当前时间。6.3 场景三批次结束标记数据流中有明确的批次结束标记如文件结束标记、批量数据的结束事件、ETL 任务的完成标记。遇到批次结束标记时发出 Watermark表示这个批次之前的数据都到齐了。优势批次级精确触发适合批流一体场景批次结束时立即触发窗口计算。示例Flink 读取文件流时每个文件结束时有 EndOfFile 事件遇到该事件发出 Watermark。6.4 场景四分区/分片结束标记数据流中有分区或分片的结束标记如 Kafka 分区的结束事件、Shard 的结束标记、分片中的高水位标记。遇到分区结束标记时发出 Watermark表示这个分区之前的数据都到齐了。优势分区级精确触发适合有界流处理分区结束时立即触发窗口计算。注意多分区场景下需要所有分区都发出 Watermark 后下游才会推进取最小值。七、断点式 Watermark 完整代码实现下面是一个完整的断点式 Watermark 实现基于事务边界事件触发 Watermarkimportorg.apache.flink.api.common.eventtime.Watermark;importorg.apache.flink.api.common.eventtime.WatermarkGenerator;importorg.apache.flink.api.common.eventtime.WatermarkOutput;importorg.apache.flink.api.common.eventtime.WatermarkStrategy;importorg.apache.flink.streaming.api.datastream.DataStream;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;importorg.apache.flink.streaming.api.windowing.time.Time;importjava.time.Duration;publicclassPunctuatedWatermarkExample{// 事件基类publicstaticclassBaseEvent{publicStringeventType;// 事件类型DATA 或 WATERMARK_MARKERpubliclongtimestamp;// 事件时间戳毫秒publicBaseEvent(){}publicbooleanisWatermarkMarker(){returnWATERMARK_MARKER.equals(eventType);}}// 普通数据事件publicstaticclassDataEventextendsBaseEvent{publicStringuserId;publicdoubleamount;publicDataEvent(){this.eventTypeDATA;}}// 事务边界标记事件publicstaticclassWatermarkMarkerEventextendsBaseEvent{publiclongtransactionId;publicWatermarkMarkerEvent(){this.eventTypeWATERMARK_MARKER;}}// 断点式 WatermarkGenerator基于事务边界事件触发publicstaticclassTransactionPunctuatedGeneratorimplementsWatermarkGeneratorBaseEvent{privatelongmaxTimestampLong.MIN_VALUE;privatelonglastEmittedWatermarkLong.MIN_VALUE;OverridepublicvoidonEvent(BaseEventevent,longeventTimestamp,WatermarkOutputoutput){// 更新最大事件时间maxTimestampMath.max(maxTimestamp,eventTimestamp);// 判断是否是事务边界标记事件if(event.isWatermarkMarker()){// 用 maxTimestamp 计算 Watermark保证单调递增longnewWatermarkmaxTimestamp;// 保证单调递增只发出比上次更大的 Watermarkif(newWatermarklastEmittedWatermark){output.emitWatermark(newWatermark(newWatermark));lastEmittedWatermarknewWatermark;}// 如果 newWatermark lastEmittedWatermark不发出被单调递增检查忽略}// 普通事件只更新 maxTimestamp不发出 Watermark}OverridepublicvoidonPeriodicEmit(WatermarkOutputoutput){// 断点式生成空实现完全由 onEvent 触发}}publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();DataStreamBaseEventsourceenv.addSource(newTransactionSource());// 断点式 Watermark 策略WatermarkStrategyBaseEventpunctuatedStrategyWatermarkStrategy.BaseEventforGenerator(ctx-newTransactionPunctuatedGenerator()).withIdleness(Duration.ofMinutes(1))// 空闲输入处理.withTimestampAssigner((event,ts)-event.timestamp);DataStreamDoubleresultsource.assignTimestampsAndWatermarks(punctuatedStrategy).keyBy(event-{if(eventinstanceofDataEvent){return((DataEvent)event).userId;}returnmarker;}).window(TumblingEventTimeWindows.of(Time.minutes(5))).sum(amount);result.print();env.execute(Punctuated Watermark Example);}}代码关键点事件类型区分用eventType字段区分普通事件“DATA”和标记事件“WATERMARK_MARKER”isWatermarkMarker()方法判断。maxTimestamp 更新每个事件到达时都更新 maxTimestamp保证 Watermark 单调递增。标记事件触发只有标记事件到达时才发出 Watermark普通事件只更新状态。单调递增保证记录 lastEmittedWatermark只发出更大的值用 maxTimestamp 而不是事件时间戳计算 Watermark。onPeriodicEmit 空实现完全由 onEvent 触发不需要周期性生成。withIdleness设置空闲输入处理避免空闲输入导致 Watermark 不推进。八、断点式 周期性组合实现兜底机制断点式 Watermark 最大的风险是标记事件丢失时 Watermark 不推进。生产环境建议断点式 周期性组合使用断点式负责事件级精确触发周期性负责兜底即使标记事件丢失也能定期推进 Watermark。// 断点式 周期性组合 WatermarkGeneratorpublicclassCombinedGeneratorimplementsWatermarkGeneratorBaseEvent{privatelongmaxTimestampLong.MIN_VALUE;privatelonglastEmittedWatermarkLong.MIN_VALUE;privatefinallongperiodicIntervalMs;// 周期性兜底间隔publicCombinedGenerator(longperiodicIntervalMs){this.periodicIntervalMsperiodicIntervalMs;}OverridepublicvoidonEvent(BaseEventevent,longeventTimestamp,WatermarkOutputoutput){maxTimestampMath.max(maxTimestamp,eventTimestamp);// 断点式标记事件到达时立即发出if(event.isWatermarkMarker()){emitWatermarkIfNeeded(output);}}OverridepublicvoidonPeriodicEmit(WatermarkOutputoutput){// 周期性兜底即使没有标记事件也定期推进 Watermark// 注意这里用 maxTimestamp - 一个小的乱序容忍时间避免提前触发if(maxTimestamp!Long.MIN_VALUE){longperiodicWatermarkmaxTimestamp-1000;// 1秒乱序容忍if(periodicWatermarklastEmittedWatermark){output.emitWatermark(newWatermark(periodicWatermark));lastEmittedWatermarkperiodicWatermark;}}}privatevoidemitWatermarkIfNeeded(WatermarkOutputoutput){longnewWatermarkmaxTimestamp;if(newWatermarklastEmittedWatermark){output.emitWatermark(newWatermark(newWatermark));lastEmittedWatermarknewWatermark;}}}组合实现的优势断点式负责精确触发标记事件到达时立即发出 Watermark事件级精确周期性负责兜底即使标记事件丢失也能定期推进 Watermark避免窗口永远不触发两者配合断点式发出的 Watermark 更精确用 maxTimestamp周期性发出的 Watermark 更保守用 maxTimestamp - 乱序容忍两者都保证单调递增九、六个常见问题9.1 问题一无触发事件时 Watermark 永远不推进现象数据流中没有标记事件或标记事件丢失Watermark 永远不推进窗口永远不触发。原因断点式 Watermark 完全由标记事件触发没有标记事件就不会调用 emitWatermark。解决方案配合周期性 Watermark 作为兜底在 onPeriodicEmit 中也发出 Watermark避免标记事件丢失时 Watermark 不推进。9.2 问题二标记事件丢失导致 Watermark 不推进现象特殊标记事件在传输过程中丢失如网络丢包、消息队列消息丢失导致 Watermark 不推进。原因标记事件是断点式的唯一触发源丢失后就没有触发了。解决方案标记事件需要可靠传输如 Kafka 的 ackall、retries0或配合周期性 Watermark 作为兜底。9.3 问题三频繁触发导致性能问题现象如果标记事件过于频繁如每条数据都是标记事件onEvent 中频繁调用 emitWatermarkWatermark 频繁向下游传播导致性能开销大。原因每个标记事件都触发一次 Watermark 发出和下游传播开销大。解决方案标记事件不要过于频繁建议间隔至少 100ms 以上或在 onEvent 中做节流只在 Watermark 真正推进时才发出即 newWatermark lastEmittedWatermark 时才发出。9.4 问题四发出的 Watermark 不单调递增现象自定义实现时如果标记事件中的时间戳回退小于当前 Watermark发出的 Watermark 不单调递增被 Flink 忽略导致 Watermark 不推进。原因标记事件的时间戳可能不是单调递增的如事务提交时间可能乱序。解决方案记录上次发出的 Watermark只发出更大的值用 maxTimestamp 计算 Watermark 而不是直接用事件时间戳maxTimestamp 保证单调递增。9.5 问题五多分区标记事件不同步现象多分区场景下不同分区的标记事件到达时间不同下游取最小值导致整体 Watermark 被慢的分区拖慢。原因多输入 Watermark 取最小值慢分区的标记事件晚到拖慢整体。解决方案合理设置并行度和分区策略保证各分区标记事件同步或配合 withIdleness 处理慢分区。9.6 问题六普通事件和标记事件混淆现象数据流中普通事件和标记事件的格式相同没有明确的区分字段导致 onEvent 中无法准确判断是否是标记事件。原因事件设计时没有区分普通事件和标记事件。解决方案标记事件需要有明确的类型字段如eventTypeWATERMARK_MARKER或用单独的事件类表示标记事件如WatermarkMarkerEvent extends BaseEvent。十、八条最佳实践优先用周期性 Watermark断点式只在有明确标记事件的特殊场景使用大多数场景用周期性 Watermark有界乱序更稳定可靠。不要为了高级而用断点式。配合周期性作为兜底断点式 Watermark 可以配合周期性 Watermark 作为兜底在 onPeriodicEmit 中也发出 Watermark避免标记事件丢失时 Watermark 不推进。标记事件可靠传输特殊标记事件需要可靠传输如 Kafka 的 ackall、retries0确保标记事件不丢失。标记事件丢失是断点式最大的风险。标记事件不要过于频繁标记事件间隔建议至少 100ms 以上避免频繁触发导致性能问题。在 onEvent 中做节流只在 Watermark 真正推进时才发出。保证单调递增记录上次发出的 Watermark只发出更大的值。用 maxTimestamp 计算 Watermark 而不是直接用事件时间戳避免标记事件时间戳回退。明确区分标记事件标记事件需要有明确的类型字段如eventTypeWATERMARK_MARKER或用单独的事件类表示避免普通事件和标记事件混淆。设置 withIdleness所有事件时间作业都应该设置withIdleness(Duration.ofMinutes(1))避免空闲输入导致 Watermark 不推进。断点式也不例外。监控 Watermark 推进监控 Current Watermark 和 Watermark Lag如果 Watermark 长时间不推进可能是标记事件丢失或上游故障及时告警排查。十一、总结与下一篇预告断点式 Watermark 是 Flink 中另一种 Watermark 生成方式要点回顾第一断点式 Watermark 在每个事件到达时调用 onEvent根据事件内容判断是否满足触发条件如果满足则立即发出 Watermark。核心特点特殊标记事件触发、事件级精确、实现复杂、无触发事件时 WM 永远不推进。第二触发时间线普通事件只更新内部状态maxTimestamp不发出 Watermark标记事件到达时立即发出 Watermark向下游传播可能触发窗口。整个过程没有固定时间间隔完全由事件内容驱动。第三内部实现三个核心组件TimestampsAndWatermarksOperator入口算子每个事件到达立即调用 onEvent、WatermarkGenerator.onEvent核心判断事件内容满足条件发出 WM、WatermarkOutput.emitWatermark输出通道单调递增检查向下游传播。第四四步生成流程事件到达→提取时间戳→调用 onEvent → onEvent 判断事件内容→普通事件只更新状态 → 满足触发条件→调用 emitWatermark 发出 WM → 单调递增检查→WM 向下游传播→驱动窗口触发。第五四大适用场景事务边界事件binlog 事务边界、交易完成事件、心跳/保活事件监控心跳包、IoT 保活、批次结束标记文件结束、ETL 完成、分区/分片结束标记Kafka 分区结束、Shard 结束。第六最大风险是无触发事件时 Watermark 永远不推进。生产环境建议断点式 周期性组合使用断点式负责事件级精确触发周期性负责兜底。第七六个常见问题无触发事件 WM 不推进、标记事件丢失、频繁触发性能问题、发出 WM 不单调递增、多分区标记不同步、普通事件和标记事件混淆。第八八条最佳实践优先用周期性、配合周期性兜底、标记事件可靠传输、标记事件不要过于频繁、保证单调递增、明确区分标记事件、设置 withIdleness、监控 Watermark 推进。