Flink 1.10.1 中使用 RocksDB 状态后端的检查点问题答案

【问题标题】：Checkpointing issues in Flink 1.10.1 using RocksDB state backendFlink 1.10.1 中使用 RocksDB 状态后端的检查点问题
【发布时间】：2020-10-01 15:22:28
【问题描述】：

我们的 Flink 工作遇到了一个非常难以观察的问题。

工作相当简单，它：

使用 Flink Kinesis 连接器从 Kinesis 读取消息
键入消息并将它们分发给大约 30 个不同的 CEP 操作员，以及几个自定义 WindowFunctions
从 CEP/Windows 发出的消息被转发到将消息写入 SQS 的 SinkFunction

我们正在运行 Flink 1.10.1 Fargate，使用 2 个 4vCPU/8GB 容器，我们使用 RocksDB 状态后端，配置如下：

state.backend: rocksdb
state.backend.async: true
state.backend.incremental: false
state.backend.rocksdb.localdir: /opt/flink/rocksdb
state.backend.rocksdb.ttl.compaction.filter.enabled: true
state.backend.rocksdb.files.open: 130048

作业以 8 的并行度运行。

当作业从冷启动时，它使用的 CPU 非常少，检查点在 2 秒内完成。随着时间的推移，检查点的大小会增加，但时间仍然非常合理，只有几秒钟：

在此期间，我们可以观察到 TaskManager 的 CPU 使用率由于某种原因而缓慢增长：

最终，检查点时间将开始飙升至几分钟，然后将开始反复超时（10 分钟）。此时：

检查点大小（完成时）约为 60MB
CPU 使用率很高，但不是 100%（通常在 60-80% 左右）
查看进行中的检查点，通常 95% 以上的操作员在 30 秒内完成检查点，但少数会坚持下去，永远不会完成。 SQS 接收器将始终包含在其中，但 SinkFunction 并不丰富且没有状态。
在这些操作员上使用背压监视器会报告高背压

最终，这种情况可以通过以下两种方式之一解决：

由于失败的检查点比例阈值，足够的检查点无法触发作业失败
检查点最终开始成功，但永远不会回到最初的 5-10 秒（当状态大小更像是 30MB 与 60MB 时）

我们真的不知道如何调试它。与您在此处的某些问题中看到的那种状态相比，我们的状态似乎非常小。我们的数据量也很低，我们经常低于 100 条记录/秒。

我们非常感谢任何关于我们可以研究调试的领域的意见。

谢谢，

【问题讨论】：

你是否使用了状态TTL，如果是，它是如何配置的？

标签： apache-flink rocksdb flink-cep

【解决方案1】：

几点：

状态随着时间的推移逐渐增长并不罕见。也许您的密钥空间正在增长，并且您为每个密钥保留了一些状态。如果您依赖状态 TTL 来使陈旧状态过期，那么它可能没有以允许它像您期望的那样快速清理过期状态的方式进行配置。无意中创建的 CEP 模式也相对容易，这些模式需要保持某种状态很长时间才能排除某些可能的匹配。

下一步是确定背压的原因。最常见的原因是工作没有足够的资源。随着时间的推移，随着（例如）用户数量的增加，大多数工作逐渐需要更多资源。例如，您可能需要增加并行度，或为实例提供更多内存，或增加接收器的容量（或网络到接收器的速度），或为 RocksDB 提供更快的磁盘。

除了供应不足之外，其他造成背压的原因还包括

正在用户函数中阻止 i/o
大量定时器同时触发
不同源之间的事件时间偏差导致大量状态被缓冲
数据偏差（热键）压倒了一个子任务或插槽
长时间的 GC 暂停
争用关键资源（例如，使用 NAS 作为 RocksDB 的本地磁盘）

启用RocksDB native metrics 可能会提供一些见解。

【讨论】：

【解决方案2】：

将此属性添加到您的配置中：

state.backend.rocksdb.checkpoint.transfer.thread.num: {threadNumberAccordingYourProjectSize}

如果你不添加这个，它将是1（默认）

链接：https://github.com/apache/flink/blob/master/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOptions.java#L62

【讨论】：