理解Flink watermark
先谈事件时间
所谓事件时间，就是Flink DataStream中的数据元素自身带有的、其实际发生时记录的时间戳，具有业务含义，并与系统时间独立。很显然，由于外部系统产生的数据往往不能及时、按序到达Flink系统，所以事件时间比处理时间有更强的不可预测性。
再谈水印
为了能够准确地表达事件时间的处理进度，就必须用到水印。Flink水印的本质是DataStream中的一种特殊元素，每个水印都携带有一个时间戳。当时间戳为T的水印出现时，表示事件时间t <= T的数据都已经到达，即水印后面应该只能流入事件时间t > T的数据。也就是说，水印是Flink判断迟到数据的标准，同时也是窗口触发的标记。 下面的图，是一个乱序的基于事件时间的数据流示例:
基于水印的数据流.png
图中的方框就是数据元素，其中的数字表示事件时间，W(x)就表示时间戳是x的水印，并有长度为4个时间单位的滚动窗口。假设时间单位为秒，可见事件时间为2、3、1s的元素都会进入区间为[1s, 4s]的窗口，而事件时间为7s的元素会进入区间为[5s, 8s]的窗口。当水印W(4)到达时，表示已经没有t <= 4s的元素了，[1s, 4s]窗口会被触发并计算。同理，水印W(9)到达时，[5s, 8s]窗口会被触发并计算，以此类推。
如果事件时间为6的元素出现在W(9)后面，就算是迟到了。后面会有迟到数据的处理。
上面的示例只有一个并行度，在有多个并行度的情况下，就会有多个流产生水印，窗口触发时该采用哪个水印呢？答案是所有流入水印中时间戳最小的那个。
来自官方文档的图能够说明问题。
多流水印.png
如果所有流入水印中时间戳最小的那个都已经达到或超过了窗口的结束时间，那么所有流的数据肯定已经全部收齐，就可以安全地触发窗口计算了。
提取事件时间、产生水印
那么事件时间是如何从数据中提取的，水印又是如何产生的呢？Flink提供了统一的DataStream.assignTimestampsAndWatermarks()方法来提取事件时间并同时产生水印。 assignTimestampsAndWatermarks()方法接受的参数类型有AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks两种，分别对应周期性水印和打点（即由事件本身的属性触发）水印，它们的类图如下所示。
周期性水印
使用AssignerWithPeriodicWatermarks时，水印是周期性产生的。该周期默认为200ms，也能通过ExecutionConfig.setAutoWatermarkInterval()方法来指定新的周期。 我们需要通过实现extractTimestamp()方法来提取事件时间，实现getCurrentWatermark()方法产生水印。但好在Flink已经提供了3种内置的实现类，所以我们直接用就可以了，省事。
- AscendingTimestampExtractor
public abstract long extractAscendingTimestamp(T element);
@Override
public final long extractTimestamp(T element, long elementPrevTimestamp) {
final long newTimestamp = extractAscendingTimestamp(element);
if (newTimestamp >= this.currentTimestamp) {
this.currentTimestamp = newTimestamp;
return newTimestamp;
} else {
violationHandler.handleViolation(newTimestamp, this.currentTimestamp);
return newTimestamp;
}
}
@Override
public final Watermark getCurrentWatermark() {
return new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - 1);
}AscendingTimestampExtractor产生的时间戳和水印必须是单调非递减的，用户通过覆写extractAscendingTimestamp()方法抽取时间戳。如果产生了递减的时间戳，就要使用名为MonotonyViolationHandler的组件处理异常，有两种方式：打印警告日志（默认）和抛出RuntimeException。
单调递增的事件时间并不太符合实际情况，所以AscendingTimestampExtractor用得不多。
- BoundedOutOfOrdernessTimestampExtractor
public BoundedOutOfOrdernessTimestampExtractor(Time maxOutOfOrderness) {
if (maxOutOfOrderness.toMilliseconds() < 0) {
throw new RuntimeException("Tried to set the maximum allowed " +
"lateness to " + maxOutOfOrderness + ". This parameter cannot be negative.");
}
this.maxOutOfOrderness = maxOutOfOrderness.toMilliseconds();
this.currentMaxTimestamp = Long.MIN_VALUE + this.maxOutOfOrderness;
}
public abstract long extractTimestamp(T element);
@Override
public final Watermark getCurrentWatermark() {
long potentialWM = currentMaxTimestamp - maxOutOfOrderness;
if (potentialWM >= lastEmittedWatermark) {
lastEmittedWatermark = potentialWM;
}
return new Watermark(lastEmittedWatermark);
}
@Override
public final long extractTimestamp(T element, long previousElementTimestamp) {
long timestamp = extractTimestamp(element);
if (timestamp > currentMaxTimestamp) {
currentMaxTimestamp = timestamp;
}
return timestamp;
}BoundedOutOfOrdernessTimestampExtractor产生的时间戳和水印是允许“有界乱序”的，构造它时传入的参数maxOutOfOrderness就是乱序区间的长度，而实际发射的水印为通过覆写extractTimestamp()方法提取出来的时间戳减去乱序区间，相当于让水印把步调“放慢一点”。这是Flink为迟到数据提供的第一重保障。
乱序区间的长度要根据实际环境谨慎设定，设定得太短会丢较多的数据，设定得太长会导致窗口触发延迟，实时性减弱。
- IngestionTimeExtractor
@Override
public long extractTimestamp(T element, long previousElementTimestamp) {
final long now = Math.max(System.currentTimeMillis(), maxTimestamp);
maxTimestamp = now;
return now;
}
@Override
public Watermark getCurrentWatermark() {
final long now = Math.max(System.currentTimeMillis(), maxTimestamp);
maxTimestamp = now;
return new Watermark(now - 1);
}IngestionTimeExtractor基于当前系统时钟生成时间戳和水印，其实就是Flink三大时间特征里的摄入时间了。
打点水印
打点水印比周期性水印用的少很多，而且Flink没有内置的实现
sourceStream.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<UserActionRecord>() {
@Nullable
@Override
public Watermark checkAndGetNextWatermark(UserActionRecord lastElement, long extractedTimestamp) {
return lastElement.getUserId().endsWith("0") ? new Watermark(extractedTimestamp - 1) : null;
}
@Override
public long extractTimestamp(UserActionRecord element, long previousElementTimestamp) {
return element.getTimestamp();
}
});AssignerWithPunctuatedWatermarks适用于需要依赖于事件本身的某些属性决定是否发射水印的情况。我们实现checkAndGetNextWatermark()方法来产生水印，产生的时机完全由用户控制。上面例子中是收取到用户ID末位为0的数据时才发射。
注意：
- 不管使用哪种方式产生水印，都不能过于频繁。因为Watermark对象是会全部流向下游的，也会实打实地占用内存，水印过多会造成系统性能下降。
- 水印的生成要尽量早，一般是在接入Source之后就产生，或者在Source经过简单的变换（map、filter等）之后产生。
- 如果需求方对事件时间的业务意义并不关心，可以直接使用处理时间，简单方便
迟到数据处理
水印的乱序区间能够保证一些迟到数据不被丢弃，但是乱序区间往往不很长，那些真正迟到了的数据该怎么办呢？一般有两种方法：
一、 窗口允许延迟
Flink提供了WindowedStream.allowedLateness()方法来设定窗口的允许延迟。即正常情况下窗口触发计算完成之后就会被销毁，但是设定了允许延迟之后，窗口会等待allowedLateness的时长后再销毁。在该区间内的迟到数据仍然可以进入窗口中，并触发新的计算。但是，窗口也是吃资源大户，所以allowedLateness的值要适当。代码示例如下：
sourceStream.assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor<UserActionRecord>(Time.seconds(30)) {
private static final long serialVersionUID = 1L;
@Override
public long extractTimestamp(UserActionRecord record) {
return record.getTimestamp();
}
}
)
.keyBy("platform")
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.seconds(30))
.aggregate(new ViewAggregateFunc(), new ViewSumWindowFunc());allowedLateness机制实际上就是DataFlow模型中的回填（backfill）策略的实现。
二、侧输出迟到数据
侧输出（side output）是Flink的分流机制。迟到数据本身可以当做特殊的流，我们通过调用WindowedStream.sideOutputLateData()方法将迟到数据发送到指定OutputTag的侧输出流里去，再进行下一步处理（比如存到外部存储或消息队列）。代码如下。
// 侧输出的OutputTag
OutputTag<UserActionRecord> lateOutputTag = new OutputTag<>("late_data_output_tag");
sourceStream.assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor<UserActionRecord>(Time.seconds(30)) {
private static final long serialVersionUID = 1L;
@Override
public long extractTimestamp(UserActionRecord record) {
return record.getTimestamp();
}
}
)
.keyBy("platform")
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.seconds(30))
.sideOutputLateData(lateOutputTag) // 侧输出
.aggregate(new ViewAggregateFunc(), new ViewSumWindowFunc())
// ......
// 获取迟到数据并写入对应Sink
stream.getSideOutput(lateOutputTag).addSink(lateDataSink);