简而言之,我想从头开始对 Kafka 中的数据重新运行 Flink 管道。
Flink 0.10.2,卡夫卡 0.8.2。
我在 Kafka 中有一个保留 2 小时的推文主题,在 Flink 中有一个管道,它以每 10 秒 5 分钟的滑动窗口计数推文。
如果我中断管道并重新运行它,我希望它重新阅读较旧的推文,从而发出 5 分钟的推文计数。相反,它似乎从新到达的推文重新开始,因此需要 5 分钟才能计数“处于状态”。
我已经尝试了两个auto.offset.reset = smallest/earliest
,并更改了group.id
,但没有成功。我还尝试按照此处所述手动更改 Kafka 中的偏移量:https ://metabroadcast.com/blog/resetting-kafka-offsets
然后我假设该问题可能与 Flink 的检查点有关,但我不知道/找不到有关如何重置它的信息。
任何人都可以分享一些工作代码吗?谢谢,E。