flink 是否保存带有水印的关闭事件时间窗口的历史记录？

Question

我有一个 flink 作业，它使用带有事件时间和水印的键控翻滚窗口来聚合数据。

我的问题是 flink 是否保留他已经关闭的窗口的状态？ 否则，我没有其他解释为什么属于以前从未打开过的窗口的事件会打开一个窗口并且不会立即删除它。

假设我们的窗口为 1 小时，forBoundedOutOfOrderness 为 10 分钟，

让我们举个例子：

event1 = ("2022-01-01T08:25:00Z") =>;窗口触发

event2 = ("2022-01-01T09:25:00Z") =>窗口已创建，但未按预期触发

event3 = ("2022-01-01T05:25:00Z") =>将与事件 4 聚合而不是丢弃（为什么？）

event4 = ("2022-01-01T05:40:00Z") =>将与事件 3 聚合而不是删除（为什么？）

    val stream = env
      .fromSource(
        kafkaSource,
        WatermarkStrategy
          .forBoundedOutOfOrderness[(String, EnrichedProcess, KafkaHeaders)](Duration.ofMinutes(outOfOrdernessMinutes)) //Watermark max time for late events
          .withIdleness(Duration.ofSeconds(idleness))
          .withTimestampAssigner(new SerializableTimestampAssigner[(String, EnrichedProcess, KafkaHeaders)] {
            override def extractTimestamp(element: (String, EnrichedProcess, KafkaHeaders), recordTimestamp: Long)
                : Long = {
              logger.info(
                LogMessage(
                  element._3.orgId,
                  s"Received incoming EnrichedProcess update_time: ${element._3.updateTime}, process time: ${recordTimestamp.asDate}",
                  element._3.flowId
                )
              )
              element._3.updateTime.asEpoch
            }
          }),
        s"Source - $kConsumeTopic"
      )

    stream
      .keyBy(element => (element._2.orgId -> element._2.procUid))                                                                     
      .window(TumblingEventTimeWindows.of(Time.hours(tumblingWindowHours), Time.minutes(windowStartingOffsetMinutes)))
      .reduce(new ReduceFunc)                                                                                         
      .name("Aggregated EnrichedProcess")
      .sinkTo(kafkaConnector.createKafkaSink(kProducerServers, kProduceTopic))
      .name(s"Sink -> $kProduceTopic")

编辑： 我测试这个的方法是使用 docker compose 进行集成测试。我正在向 Kafka => 生成一个事件Flink job 消耗的沉入Kafka =>检查kafka的内容。

当我在发送事件之间设置 30 秒的睡眠时，事件 3 和事件 4 被丢弃。这是我所期待的行为。

    val producer = new Producer(producerTopic)

    val consumer = new Consumer(consumerTopic, groupId)
    producer.send(event1)
    producer.send(event2)
    Thread.sleep(30000)
    producer.send(event3)
    Thread.sleep(30000)
    producer.send(event4)

    val received: Iterable[(String, EnrichedProcess)] = consumer.receive[EnrichedProcess]()

但现在更好奇的是，为什么当我设置 10 秒而不是 30 秒的睡眠时，我只收到第一种情况（水印应该已经更新（定期水印生成器的默认值为 200 毫秒）

Answer 1

执行摘要：

Flink 基于事件时间的逻辑中的非确定性来自于处理时间与事件时间的混合——就像周期性水印生成器和空闲检测所发生的那样。只有当您永远不会有迟到的事件或闲置的资源时，您才能确定确定性的结果。

更多详细信息：

虽然您预计

event3 = ("2022-01-01T05:25:00Z")

会迟到，但只有在足够大的水印成功先到达的情况下，才会真正迟到。使用forBoundedOutOfOrderness策略并不能保证这一点——这是一个周期性水印生成器，每200毫秒生成一次水印。因此，可能在 event3 和 event4 之间产生基于 event2 时间戳的水印。

这是一种可能的解释；根据具体情况，可能还有其他情况。例如，随着所有睡眠的进行，水印生成器的一个并行实例至少空闲一分钟，这可能涉及产生被观察到的行为（取决于idleness 等）。

更多背景：

并行度> 1、水印策略有多个独立的实例，每个实例根据它们处理的事件做自己的事情。

具有多个输入通道（例如键控窗口）的操作员将通过将传入水印（来自所有非空闲通道）的最小值作为自己的水印来组合这些水印。

回答原来的问题：

Flink 是否保留已经关闭的窗口的状态？ 不会。一旦允许的延迟（如果有）过期，事件的状态时间窗口被清除。