Flink WaterMark与触发器疑问：事件时间下晚到元素未被丢弃？

嗨，我懂你现在对Flink事件时间水印处理晚到元素的逻辑有点懵，咱们先把你理解里的小偏差掰正，再结合常见示例理清楚～

• 水印推进不是“见大就更”：你以为只要遇到事件时间比当前水印大的数据就会推进水印？其实不是的！水印的生成逻辑完全取决于你用的水印生成器：
  • 比如常用的BoundedOutOfOrdernessTimestampExtractor，是拿当前处理到的最大事件时间减去你设置的乱序容忍时间，得到新的水印——只有当这个新水印比当前的水印大时，才会推进；
  • 如果是自定义水印生成器，那逻辑就更灵活了，甚至可能不是每条数据都触发水印更新。
• 窗口触发的条件是“窗口结束时间 ≤ 当前水印”：你说的“严格小于”是错的哦，而且默认情况下，窗口触发后不会立刻被清理——除非你没设迟到容忍时间，且后续的水印超过了窗口结束时间+迟到容忍（默认是0），窗口才会被彻底清理。

假设你的未写完的代码是类似这样的典型场景：

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor

object FlinkLateEventDemo {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // Flink 1.12+ 已经默认是EventTime，这里写出来是为了明确
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 模拟数据流：(key, 内容, 事件时间戳(ms))
    val data = env.fromElements(
      (1, "a", 1000L),  // 事件时间1s
      (1, "b", 3000L),  // 事件时间3s
      (1, "c", 2000L),  // 晚到的事件，时间2s
      (1, "d", 5000L)   // 事件时间5s
    )

    // 分配时间戳和水印：设置1s的乱序容忍
    val stream = data.assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[(Int, String, Long)](Time.seconds(1)) {
        override def extractTimestamp(element: (Int, String, Long)): Long = element._3
      }
    )

    // 按key分组，开3s的滚动窗口，做reduce聚合
    stream.keyBy(_._1)
      .timeWindow(Time.seconds(3))  // 窗口区间是[0,3), [3,6), ...
      .reduce((a,b) => (a._1, a._2 + "," + b._2, Math.max(a._3, b._3)))
      .print()

    env.execute()
  }
}

咱们一步步看这个例子里的水印和窗口变化：

1. 处理第一条数据(1,"a",1000L)：最大事件时间是1000L，水印=1000-1000=0L。此时窗口[0,3)的结束时间是3000L，比水印大，不触发计算。
2. 处理第二条数据(1,"b",3000L)：最大事件时间更新为3000L，水印=3000-1000=2000L。窗口结束时间3000L还是大于水印，继续等。
3. 处理第三条晚到数据(1,"c",2000L)：最大事件时间还是3000L，新水印=3000-1000=2000L，和当前水印一样，所以水印不推进。窗口依然不触发。
4. 处理第四条数据(1,"d",5000L)：最大事件时间更新为5000L，水印=5000-1000=4000L。此时窗口[0,3)的结束时间3000L ≤ 4000L，触发窗口计算，输出结果(1,a,b,c,3000)。

• 如果给窗口加上allowedLateness(Time.seconds(2))，窗口[0,3)在水印到4000L时触发第一次计算，但会一直保留到水印达到3000+2000=5000L时才彻底清理。在这期间，任何事件时间属于[0,3)的晚到数据（比如时间2500L的元素）还能触发窗口的增量更新。
• 如果没设置allowedLateness（默认容忍0秒），窗口触发后，后续再进来的事件时间远小于水印的元素会被直接丢弃，窗口状态也会被清理。
• 要是不想丢弃晚到元素，还可以用sideOutputLateData把这些元素导到侧输出流里，方便后续做兜底处理。

内容的提问来源于stack exchange，提问作者GeorgeWilson