Flink Kinesis Consumer未存储上次成功处理的序列号

Question

我们正在使用Flink Kinesis消费者将Kinesis流中的数据消费到我们的Flink应用程序中。

KCL库使用DynamoDB表存储上次成功处理的Kinesis流序列号。这样，下次应用程序启动时，它将从停止的位置恢复。

但是，Flink Kinesis消费者似乎在任何持久性存储中都不维护任何这样的序列号。因此，我们需要依赖ShardIteratortype（trim\u horizen、latest等）来决定在应用程序重新启动时恢复Flink应用程序处理的位置。

一个可能的解决方案是依赖Flink检查点机制，但这仅在应用程序失败后恢复时有效，而不是在应用程序被故意取消并且需要从上次成功使用的Kinesis流序列号重新启动时有效。

我们需要自己存储这些最后成功消耗的序列号吗？

Answer 1

为了补充David的回答，我想解释一下不存储序列号背后的原因。

提交到源系统中的任何类型的偏移量都会将检查点/保存点功能限制在容错范围内。也就是说，只有最新的检查点/保存点才能恢复。

然而，Flink实际上支持跳回以前的检查点/保存点。考虑升级应用程序。您可以在升级之前创建一个保存点，并让它运行几分钟，在那里创建几个检查点。然后，您会发现一个关键的bug。您希望回滚到已获取的保存点并放弃所有检查点。

现在，如果Flink只将源偏移提交给源系统，那么从现在起到恢复的保存点之间，我们将无法重播数据。因此，正如David指出的那样，Flink需要将偏移量存储在保存点本身中。此时，额外提交到源系统并不会带来任何好处，而且在恢复到以前的保存点/检查点时会造成混乱。

您认为额外存储偏移有什么好处吗？

Answer 2

Flink的最佳实践是使用检查点和保存点，因为这些检查点和保存点会创建一致的快照，其中包含消息队列中的偏移量（在本例中，是运动流序列号），以及作业图其余部分中的所有状态，这些状态是由于消耗了这些偏移量之前的数据。这使恢复或重新启动而不丢失或重复数据成为可能。

Flink的检查点是Flink为从故障中恢复而自动拍摄的快照，其格式针对快速恢复进行了优化。保存点使用相同的底层快照机制，但手动触发，其格式更关注操作灵活性而非性能。

保存点是您需要的。特别是，使用保存点取消和从保存点恢复非常有用。

另一个选项是将保留的检查点与ExternalizedCheckpointCleanup一起使用。RETAIN\u ON\u取消。