Flink 如何使用来自具有多个分区的 Kafka 主题的消息，而不会产生偏差?

Question

假设我们有一个主题的 3 个 kafka 分区，我希望我的事件使用事件时间按小时窗口。

当分区位于当前窗口之外时，kafka 使用者是否会停止从分区读取？还是打开一个新窗口？如果它正在打开新窗口，那么理论上是否可能让它打开无限数量的窗口，从而耗尽内存，如果一个分区的事件时间与其他分区相比非常偏差？当我们重播一些历史时，这种情况尤其可能。

我一直试图通过阅读文档来获得这个答案，但找不到太多关于 Flink 与 Kafka 在分区上的内部信息。非常欢迎有关此特定主题的一些良好文档。

谢谢！

Answer 1

因此，首先，来自 Kafka 的事件会不断被读取，进一步的窗口操作对此没有影响。在谈论内存不足时，还有更多需要考虑的事情。

• 通常，您不会存储窗口的每个事件，而只是存储事件的一些聚合
• 每当窗口关闭时，都会释放相应的内存。

更多关于Kafka消费者如何与EventTime交互的信息(特别是水印，你可以在这里查看。

Answer 2

您可以尝试使用这种类型的样式

public void runStartFromLatestOffsets() throws Exception {
// 50 records written to each of 3 partitions before launching a latest-starting consuming job
final int parallelism = 3;
final int recordsInEachPartition = 50;
// each partition will be written an extra 200 records
final int extraRecordsInEachPartition = 200;
// all already existing data in the topic, before the consuming topology has started, should be ignored
final String topicName = writeSequence("testStartFromLatestOffsetsTopic", recordsInEachPartition, parallelism, 1);
// the committed offsets should be ignored
KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
kafkaOffsetHandler.setCommittedOffset(topicName, 0, 23);
kafkaOffsetHandler.setCommittedOffset(topicName, 1, 31);
kafkaOffsetHandler.setCommittedOffset(topicName, 2, 43);