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可能会导致两个窗口合并成一个窗口。

Flink时间系列文章：