Event Time语义下我们使用Watermark来判断数据是否迟到。一个迟到元素是指元素到达窗口算子时，该元素本该被分配到某个窗口，但由于延迟，窗口已经触发计算。目前Flink有三种处理迟到数据的方式：
- 直接将迟到数据丢弃
- 将迟到数据发送到另一个流
- 重新执行一次计算，将迟到数据考虑进来，更新计算结果
将迟到数据丢弃
如果不做其他操作，默认情况下迟到数据会被直接丢弃。
将迟到数据发送到另外一个流
如果想对这些迟到数据处理，我们可以使用Flink的侧输出（Side Output）功能，将迟到数据发到某个特定的流上。后续我们可以根据业务逻辑的要求，对迟到的数据流进行处理。
// 数据流有三个字段：（key, 时间戳, 数值）
val input: DataStream[(String, Long, Int)] = ...
val mainStream = input.keyBy(item => item._1)
.timeWindow(Time.seconds(5))
// 将输出写到late-elements里
.sideOutputLateData(new OutputTag[(String, Long, Int)]("late-elements"))
.aggregate(new CountAggregate)
// 接受late-elements，形成一个数据流
val lateStream: DataStream[(String, Long, Int)] = mainStream.getSideOutput(new OutputTag[(String, Long, Int)]("late-elements"))
上面的代码将迟到的内容写进名为“late-elements”的OutputTag下，之后使用getSideOutput获取这些迟到的数据。
更新计算结果
对于迟到数据，使用上面两种方法，都对计算结果的正确性有影响。如果将数据流发送到单独的侧输出，我们仍然需要完成单独的处理逻辑，相对比较复杂。更理想的情况是，将迟到数据重新进行一次，得到一个更新的结果。
allowedLateness允许用户在Event Time下对某个窗口先得到一个结果，如果在一定时间内有迟到数据，迟到数据会和之前的数据一起重新被计算，以得到一个更准确的结果。使用这个功能时需要注意，原来窗口中的状态数据在窗口已经触发的情况下仍然会被保留，否则迟到数据到来后也无法与之前数据融合。另一方面，更新的结果要以一种合适的形式输出到外部系统，或者将原来结果覆盖，或者同时保存且有时间戳以表明来自更新后的计算。比如，我们的计算结果是一个键值对（Key-Value），我们可以把这个结果输出到Redis这样的KV数据库中，使用某些Reids命令，对于同一个Key下，旧的结果被新的结果所覆盖。
如果不明确调用allowedLateness，默认的允许延迟的参数是0。如果对一个Processing Time下的程序使用allowedLateness，将引发异常。
// ProcessWindowFunction接收的泛型参数分别为：[输入类型、输出类型、Key、Window]
class AllowedLatenessFunction extends ProcessWindowFunction[
(String, Long, Int), (String, String, Int, String), String, TimeWindow] {
override def process(key: String,
context: Context,
elements: Iterable[(String, Long, Int)],
out: Collector[(String, String, Int, String)]): Unit = {
// 是否被迟到数据更新
val isUpdated = context.windowState.getState(
new ValueStateDescriptor[Boolean]("isUpdated", Types.of[Boolean])
)
val count = elements.size
val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
if (isUpdated.value() == false) {
// 第一次使用process函数时， Boolean默认初始化为false，因此窗口函数第一次被调用时会进入这里
out.collect((key, format.format(Calendar.getInstance().getTime), count, "first"))
isUpdated.update(true)
} else {
// 之后isUpdated被置为true，窗口函数因迟到数据被调用时会进入这里
out.collect((key, format.format(Calendar.getInstance().getTime), count, "updated"))
}
}
}
// 数据流有三个字段：（key, 时间戳, 数值）
val input: DataStream[(String, Long, Int)] = ...
val allowedLatenessStream = input.keyBy(item => item._1)
.timeWindow(Time.seconds(5))
.allowedLateness(Time.seconds(5))
.process(new AllowedLatenessFunction)
在上面的代码中，我们设置的窗口为5秒，5秒结束后，窗口计算会被触发，生成第一个计算结果。allowedLateness设置窗口结束后还要等待长为lateness的时间，某个迟到元素的Event Time大于窗口结束时间但是小于结束时间+lateness，该元素仍然会被加入到该窗口中。每新到一个迟到数据，迟到数据被加入ProcessWindowFunction的缓存中，窗口的Trigger会触发一次FIRE，窗口函数被重新调用一次，计算结果得到一次更新。
需要注意的是，会话窗口依赖Session gap来切分窗口，使用了allowedLateness可能会导致两个窗口合并成一个窗口。
本文所涉及代码已经更新至github：github.com/luweizheng/…
Flink时间系列文章：
