FlinkKafkaConsumer中事件时间顺序的保证

Question

TL；DR：目前保证Flink中事件时间顺序的最佳解决方案是什么？

我使用Flink 1.8.0和Kafka 2.2.1。我需要通过事件时间戳保证事件的正确顺序。我每隔1秒生成周期性水印。我使用Flink Kafka消费者与AscendingTimestampExtractor：

java prettyprint-override">val rawConsumer = new FlinkKafkaConsumer[T](topicName, deserializationSchema, kafkaConsumerConfig)
  .assignTimestampsAndWatermarks(new AscendingTimestampExtractor[T] {
      override def extractAscendingTimestamp(element: T): Long =
        timestampExtractor(element)
      })
 .addSource(consumer)(deserializationSchema.getProducedType).uid(sourceId).name(sourceId)

然后处理：

myStream
   .keyBy(ev => (ev.name, ev.group))
   .mapWithState[ResultEvent, ResultEvent](DefaultCalculator.calculateResultEventState)

我意识到，对于在同一毫秒或几毫秒之后发生的无序事件，Flink不会纠正顺序。我在文档中发现：

水印触发所有窗口的计算，其中最大时间戳（结束时间戳-1）小于新水印

https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/windows.html#interaction-水印和窗口的应用

所以我准备了额外的处理步骤来保证事件时间顺序：

myStream
      .timeWindowAll(Time.milliseconds(100))
      .apply((window, input, out: Collector[MyEvent]) => input
        .toList.sortBy(_.getTimestamp)
        .foreach(out.collect) // this windowing guarantee correct order by event time
      )(TypeInformation.of(classOf[MyEvent]))
      .keyBy(ev => (ev.name, ev.group))
      .mapWithState[ResultEvent, ResultEvent](DefaultScoring.calculateResultEventState)

然而，我发现这个解决方案很难看，它看起来像一个变通方法。我还担心KafkaSource的每分区水印

理想情况下，我希望将顺序保证放在KafkaSource中，并为每个kafka分区保留它，就像每个分区的水印一样。有可能吗？目前保证Flink事件时间顺序的最佳解决方案是什么？

Answer 1

这是一个伟大的观点。在KafkaSource中保证秩序实际上包括两个部分。

1. 保证同一子任务中分区之间的顺序。
2. 保证子任务之间的顺序。

第一部分已经在进行中https://issues.apache.org/jira/browse/FLINK-12675.第二部分需要子任务之间共享状态的支持，这可能需要社区中更多的讨论和详细的计划。

回到你的问题，我认为通过设置缓冲数据的窗口来保持事件的顺序是目前最好的解决方案。

Answer 2

Flink不保证按事件时间顺序处理记录。分区内的记录将按其原始顺序进行处理，但当两个或多个分区合并到一个新分区时（由于流的重新分区或合并），Flink会将这些分区的记录随机合并到新分区中。其他一切都将是低效的，并导致更高的延迟。

例如，如果作业有一个从两个Kafka分区读取的源任务，两个分区的记录将以某种随机的锯齿形模式合并。

但是，Flink保证针对生成的水印正确处理所有事件。这意味着，水印永远不会超过记录。例如，如果Kafka源生成每个分区的水印，即使合并了多个分区的记录，水印仍然有效。水印用于收集和处理时间戳小于水印的所有记录。因此，它确保输入数据的完整性。

这是按时间戳对记录进行排序的先决条件。你可以用一个翻滚的窗户来做这一切。但是，您应该知道

1. 所有窗口都将在单个任务中执行（即，它不是并行的）。如果每个键的顺序足够，您应该使用一个常规的翻滚窗口，或者更好地实现一个更高效的KeyedProcessFunction