Flink 检查点不断失败答案

【问题标题】：Flink checkpoints keeps failingFlink 检查点不断失败
【发布时间】：2021-01-27 06:29:36
【问题描述】：

我们正在尝试使用 RocksDB 后端设置 Flink 有状态作业。我们正在使用会话窗口，间隔 30 分钟。我们使用 aggregateFunction，所以不使用任何 Flink 状态变量。通过采样，我们每秒有不到 20k 个事件，每秒有 20 - 30 个新会话。我们的会议基本上收集了所有的事件。会话累加器的大小会随着时间的推移而增加。我们在 Flink 1.9、128 个容器中总共使用了 10G 内存。以下是设置：

state.backend: rocksdb
state.checkpoints.dir: hdfs://nameservice0/myjob/path
state.backend.rocksdb.memory.managed: true
state.backend.incremental: true
state.backend.rocksdb.memory.write-buffer-ratio: 0.4
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1

containerized.heap-cutoff-ratio: 0.45
taskmanager.network.memory.fraction: 0.5
taskmanager.network.memory.min: 512mb
taskmanager.network.memory.max: 2560mb

根据我们对给定时间的监控， RocksDB memtable 大小小于 10m，我们的堆使用量小于 1G，但我们的直接内存使用量（网络缓冲区）使用的是 2.5G。缓冲池/缓冲区使用指标都为 1（已满）。我们的检查站不断失败，不知道网络缓存部分会占用这么多内存是正常的吗？

如果您能提供一些建议，我将不胜感激:) 谢谢！

【问题讨论】：

标签： apache-flink flink-streaming

【解决方案1】：

不管怎样，会话窗口确实在内部使用了 Flink 状态。（大多数源和接收器也是如此。）根据您将会话事件收集到会话累加器中的方式，这可能是一个性能问题。如果您需要将所有事件收集在一起，为什么要使用 AggregateFunction 执行此操作，而不是让 Flink 为您执行此操作？

为了获得最佳的窗口化性能，您希望使用 ReduceFunction 或 AggregateFunction 来逐步减少/聚合窗口，只保留最终将成为窗口结果的一小部分状态。另一方面，如果你只使用没有预聚合的 ProcessWindowFunction，那么 Flink 将在内部使用一个附加列表状态对象，当与 RocksDB 一起使用时非常有效——它只需要序列化每个事件以将其附加到末尾的名单。当窗口最终被触发时，列表将作为一个 Iterable 交付给您，该 Iterable 以块的形式反序列化。另一方面，如果您使用 AggregateFunction 推出自己的解决方案，您可能会在每次访问/更新时让 RocksDB 反序列化和重新序列化累加器。这可能会变得非常昂贵，并且可以解释检查点失败的原因。

您分享的另一个有趣事实是缓冲池/缓冲区使用指标显示它们已被充分利用。这表明存在显着的背压，这反过来又可以解释检查点失败的原因。检查点依赖于检查点屏障能够遍历整个执行图，对每个操作员进行检查点，并在超时之前完成作业的完整扫描。使用背压，这可能会失败。

背压的最常见原因是配置不足 - 或者换句话说，集群不堪重负。由于运营商跟不上，网络缓冲池被充分利用。答案不是增加缓冲，而是消除/修复瓶颈。

【讨论】：

非常感谢大卫，我切换到 ProcessWindowFunction，现在检查点都成功了：) 但是网络缓冲区使用率仍然很高（比以前好），我会更新图表中的问题本身。我检查了我们的工作，但 Flink UI 上没有显示背压。对于 RocksDB 的使用情况，指标显示使用的内存比我在 Flink UI 上看到的状态大小要少得多。你能推荐一下吗？
在 Flink 1.9 中，背压检测不是很健壮。缓冲池和 inputQueueLength 指标讲述了一个更完整的故事——看起来你有背压，但它不是连续的。在这一点上，您的第一大顾虑是什么？您是否尝试过增加并行度？
我担心的是，如果现在网络缓冲区的使用已经满了，那么当流量增长或出现一些峰值时，该作业是否能够处理。现在我们正在使用大约 1/15 的流量进行测试。我们有 128 个容器，现在每个容器总共有 20G 内存，CPU 使用率非常低，通常不到 1%。我们可能会尝试使用 256 个容器（因为我们的 Kafka 源有 256 个分区），但感觉就像在浪费 CPU 资源，并且更多的机器相互通信，可能会引入网络瓶颈。
理想情况下，您应该配置该作业，以便它可以处理正常负载，只有偶尔的背压，这样您就有能力处理典型的负载峰值而不会落后（假设您的目标是“跟上实时”场景）。（顺便说一句，低 CPU 可能会产生误导——由于数据倾斜，您可能有一个热键压倒一个核心，而其余核心什么也不做。）
很难通过 SO 提供好的建议；大多只能回答直截了当的问题。但是我们有时会做的一件事是实现一个参数化的数据生成器，该生成器近似于生产数据的显着特征，然后使用它来查看管道在不同类型的负载下如何反应。有关如何检查您的工作是否健康的一些建议，请参阅 flink.apache.org/news/2019/02/25/monitoring-best-practices.html。