TL; DR: Какое на данный момент лучшее решение для гарантии порядка событий во времени во Flink?
Я использую Flink 1.8.0 с Kafka 2.2.1. Мне нужно гарантировать правильный порядок событий по метке времени события. Я генерирую периодические водяные знаки каждые 1 с. Я использую FlinkKafkaConsumer с AscendingTimestampExtractor:
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-of-watermarks-and-windows
Итак, я подготовил дополнительный шаг обработки, чтобы гарантировать порядок событий:
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, например водяные знаки на разделы. Возможно ли это сделать? Какое на данный момент лучшее решение для гарантии порядка событий во времени во Флинке?