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

Question

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

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

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

Answer 1

当您说您使用“内部带有过滤器和多个运算符的迭代循环”时，您是否在流作业中使用 Flink 的迭代结构？

不建议这样做。正如 documentation 中所述:

Flink currently only provides processing guarantees for jobs without iterations. Enabling checkpointing on an iterative job causes an exception. In order to force checkpointing on an iterative program the user needs to set a special flag when enabling checkpointing: env.enableCheckpointing(interval, CheckpointingMode.EXACTLY_ONCE, force = true).

Please note that records in flight in the loop edges (and the state changes associated with them) will be lost during failure.

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