自定义 Watermark生成器，需要实现WatermarkGenerator接口。
/**
* 可以基于事件或者周期性的生成 watermark。
*/
@Public
public interface WatermarkGenerator<T> {
/**
* 每来一条事件数据调用一次，可以检查或者记录事件的时间戳，或者也可以基于事件数据本身去生成 watermark。
*/
void onEvent(T event, long eventTimestamp, WatermarkOutput output);
/**
* 周期性的调用，也许会生成新的 watermark，也许不会。
*
* <p>调用此方法生成 watermark 的间隔时间由 {@link ExecutionConfig#getAutoWatermarkInterval()} 决定。
*/
void onPeriodicEmit(WatermarkOutput output);
}watermark 的生成方式本质上是有两种：周期性生成和标记生成。
周期性生成器通常通过 onEvent() 观察传入的事件数据，然后在框架调用 onPeriodicEmit() 时发出 watermark。
标记生成器将查看 onEvent() 中的事件数据，并等待检查在流中携带 watermark 的特殊标记事件或打点数据。当获取到这些事件数据时，它将立即发出 watermark。通常情况下，标记生成器不会通过 onPeriodicEmit() 发出 watermark。
自定义周期性 Watermark 生成器
周期性生成器会观察流事件数据并定期生成 watermark。
其生成可能取决于流数据，或者完全基于处理时间。这是什么意思呢？
这是生成两种watermark的方式。“可能取决于流数据”，也就是在onEvent方法中基于每条数据计算一个较新时间戳，这个时间戳就会在onPeriodicEmit方法中发出新的watermark。
“完全基于处理时间”，则是在onEvent方法中什么都不做，只是在onPeriodicEmit方法中按照方法触发的时间节点，每次都生成一个新的watermark。
生成 watermark 的时间间隔（每 n 毫秒）可以通过 ExecutionConfig.setAutoWatermarkInterval(…) 指定。每次都会调用生成器的 onPeriodicEmit() 方法，如果返回的 watermark 非空且值大于前一个 watermark，则将发出新的 watermark。
如下是两个使用周期性 watermark 生成器的简单示例。注意：Flink 已经附带了 BoundedOutOfOrdernessWatermarks，它实现了 WatermarkGenerator，其工作原理与下面的 BoundedOutOfOrdernessGenerator 相似。可以在这里参阅如何使用它的内容。
/**
* 该 watermark 生成器可以覆盖的场景是：数据源在一定程度上乱序。
* 即某个最新到达的时间戳为 t 的元素将在最早到达的时间戳为 t 的元素之后最多 n 毫秒到达。
*/
public class BoundedOutOfOrdernessGenerator implements WatermarkGenerator<MyEvent> {
private final long maxOutOfOrderness = 3500; // 3.5 秒
private long currentMaxTimestamp;
@Override
public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
currentMaxTimestamp = Math.max(currentMaxTimestamp, eventTimestamp);
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
// 发出的 watermark = 当前最大时间戳 - 最大乱序时间
output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness - 1));
}
}/**
* 该生成器生成的 watermark 滞后于处理时间固定量。它假定元素会在有限延迟后到达 Flink。
*/
public class TimeLagWatermarkGenerator implements WatermarkGenerator<MyEvent> {
private final long maxTimeLag = 5000; // 5 秒
@Override
public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
// 处理时间场景下不需要实现
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
output.emitWatermark(new Watermark(System.currentTimeMillis() - maxTimeLag));
}
}自定义标记 Watermark 生成器
标记 watermark 生成器观察流事件数据并在获取到带有 watermark 信息的特殊事件元素时发出 watermark。也就是在每个元素进入时（onEvent方法触发），都会根据规则，生成watermark。
如下是实现标记生成器的方法，当事件带有某个指定标记时，该生成器就会发出 watermark：
public class PunctuatedAssigner implements WatermarkGenerator<MyEvent> {
@Override
public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
if (event.hasWatermarkMarker()) {
output.emitWatermark(new Watermark(event.getWatermarkTimestamp()));
}
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
// onEvent 中已经实现
}
}注意： 可以针对每个事件去生成 watermark。但是由于每个 watermark 都会在下游做一些计算，因此过多的 watermark 会降低程序性能。
我们总结一下两种生成方式的区别：
自定义周期性生成Watermark，onEvent方法是可选的，里面可以做逻辑处理，也可以什么都不做，最终输出Watermark，是通过onPeriodicEmit方法。
自定义标记生成Watermark，onEvent直接出数Watermark，onPeriodicEmit方法什么都不做。