生成时机
水位线生产的最佳位置是在尽可能靠近数据源的地方，因为水位线生成时会做出一些有关元素顺序相对时间戳的假设。
由于数据源读取过程是并行的，一切引起Flink跨行数据流分区进行重新分发的操作（比如：改变并行度，keyby 等）都会导致元素时间戳乱序。
但是如果是某些初始化的 filter、map 等不会引起元素重新分发的操作，可以考虑在生成水位线之前使用。
分配器
Periodic Watermarks
周期性分配水位线比较常用，是我们会指示系统 以固定的时间间隔发出的水位线。在设置时间为事件时间时，会默认设置这个时间间隔为 200ms, 如果需要调整可以自行设置。
比如下面的例子是手动设置每隔 1s 发出水位线：
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// 手动设置时间间隔为1s
env.getConfig().setAutoWatermarkInterval(1000);
周期水位线需要实现接口：AssignerWithPeriodicWatermarks，下面是示例：
public class TestPeriodWatermark implements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {
Long currentMaxTimestamp = 0L;
final Long maxOutOfOrderness = 1000L;// 延迟时长是1s
@Nullable
@Override public Watermark getCurrentWatermark() {
return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
}
@Override
public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
long timestamp = element.f1;
currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
return timestamp;
}
}
Punctuated Watermarks
定点水位线 不常用，主要为输入流中包含一些用于指示系统进度的特殊元组和标记，方便根据输入元素生成水位线的场景使用的。
由于数据流中每一个递增的EventTime都会产生一个Watermark。
在实际的生产中Punctuated方式在TPS很高的场景下会产生大量的Watermark在一定程度上对下游算子造成压力，所以只有在实时性要求非常高的场景才会选择Punctuated的方式进行Watermark的生成。
参考：
https://www.jianshu.com/p/c52ad3f284ee
https://zhuanlan.zhihu.com/p/342271617