如何修复 Apache Flink 中卡住的检查点

Question

我在 Flink 1.7.2 中有一个运行在 Cloudera 管理的集群（通过 Yarn 分配资源）上的设置，它从外部 Kafka 获取大量数据，并通过一系列聚合、计算、再次聚合的操作符进行管道传输...我什至使用了一个内部带有过滤器和多个运算符的迭代循环，最后是一个将结果写入我的 Hadoop 集群上的 RocksDB 后端的接收器。所有这些都工作了一段时间（目前大约 2-3 小时），然后检查点卡住了。 我使用完全一次检查点，检查点之间有 30 分钟的超时和 10 分钟的暂停。 1 个并发检查点。只要一切正常，这些检查点会在 1 分钟内完成。但是几个小时后，一个检查点卡住了，这意味着 Checkpoint-UI 选项卡告诉我一个（或多个）操作员尚未确认所有子任务。到那时，正常的过程也将陷入困境。我的输入源上的水印不会继续，也不会产生更多的输出。他们不会直到计时器用完。然后下一个检查点立即激活，写入可能所有任务的 10% 并再次卡住。没有恢复的机会。如果我取消作业并以最后一个成功的检查点为起点重新启动它，下一个检查点将以同样的方式卡住。

我已经尝试了很多不同的方法，从更改检查点频率到超时。我什至从完全一次更改为至少一次，因为对齐缓冲有时会变得非常昂贵。但即便如此，同样的问题也在同样的时间后出现。 资源分配似乎也没有发挥作用，我目前每个任务管理器使用 4 个任务槽，并且不时更改管理器的数量，但没有任何变化。 JVM 堆大小似乎也不是问题，因为我提交了多个 GB，但显然只使用了几百 MB。

作业或任务管理器没有发出错误消息，所有日志告诉我的是尝试写入检查点、丢失的成功消息以及下一个检查点的开始。

Answer 1

当你说你使用“内部有过滤器和多个操作符的迭代循环”时，你是在使用 Flink 的迭代构造和流式作业吗？

不建议这样做。正如documentation 中所说：

Flink 目前只为没有迭代的作业提供处理保证。在迭代作业上启用检查点会导致异常。为了在迭代程序上强制检查点，用户需要在启用检查点时设置一个特殊标志：env.enableCheckpointing(interval, CheckpointingMode.EXACTLY_ONCE, force = true)。

请注意，循环边缘中的飞行记录（以及与之相关的状态更改）将在失败期间丢失。

也就是说，您所描述的情况听起来像是背压阻止了检查点障碍的进展。很多事情都可能导致这种情况，但this blog post 可能会帮助您诊断问题。但我不确定其中有多少适用于使用迭代的工作。

Answer 2

请考虑管道中的数据偏差~，也许您可​​以增加处理器的并行度以使其更加平衡