使用RocksDB状态后端的Flink 1.10.1中的检查点问题
我们的Flink工作遇到了一个很难观察的问题.
这份工作相当简单，它：
- 使用Flink Kinesis连接器从Kinesis读取消息
- 键入消息，并将它们分发给~30个不同的WindowFunctions操作符，外加几个自定义的CEP操作符
- 从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秒内完成。随着时间的推移，检查点的大小会增加，但时间仍然是非常合理的--几秒钟：
在此期间，我们可以观察到由于某种原因，我们的TaskManagers的CPU使用量在缓慢增长：
最终，检查点时间将开始增加到几分钟，然后只会重复启动，超时 (10分钟)。此时：
- 检查点大小(完成时)约为60 is。
- CPU使用率很高，但不是100% (通常在60-80%左右)。
- 查看正在进行的检查点，通常操作符的95%+会在30秒内完成检查点，但是有几个检查点只会坚持，而且永远不会完成。SQS接收器将始终包含在这上面，但是
SinkFunction并不丰富，也没有状态。 - 使用背压监视器对这些操作者报告一个高背压。
最终，这种情况解决了以下两种方法之一：
- 由于检查点比例阈值失败，足够多的检查点无法触发作业失败。
- 检查点最终开始成功，但永远不要回到最初使用的5-10秒(当状态大小更像是30 to，而不是60 to)。
我们真的不知道如何调试它。与你在这里的一些问题中看到的那种状态相比，我们的状态似乎很小。我们的音量也很低，我们经常低于100条记录/秒。
我们将非常感谢任何关于我们可以查看的领域的输入来调试这一点。
谢谢,
回答 2
Stack Overflow用户
发布于 2020-10-02 09:03:26
以下几点：
随着时间的推移，国家逐渐增长是很正常的。也许您的密钥空间正在增长，并且您正在为每个密钥保留一些状态。如果您依赖状态TTL来过期陈旧状态，那么它的配置方式可能不会像您预期的那样快速清除过期状态。另外，很容易无意中创建CEP模式，这些模式需要在很长时间内保持某些状态，然后才能排除某些可能的匹配。
下一步很好的办法是找出产生背压的原因。最常见的原因是一份工作没有足够的资源。随着时间的推移，大多数工作逐渐需要更多的资源，因为被管理的用户数量(例如)增加了。例如，您可能需要增加并行性，或者给实例更多的内存，或者增加接收器的容量(或者网络对接收器的速度)，或者给RocksDB更快的磁盘。
除了供给不足外，造成背压的其他原因还包括
- 阻塞i/o是在用户函数中完成的。
- 大量计时器同时开火。
- 不同源之间的事件时间偏差导致大量状态被缓冲。
- 数据倾斜(热键)是压倒一个子任务或时隙。
- 长时间GC暂停
- 关键资源的争用(例如，使用NAS作为RocksDB的本地磁盘)
启用RocksDB本机度量可能会提供一些见解。
Stack Overflow用户
发布于 2020-10-07 10:00:16
将此属性添加到配置中：
state.backend.rocksdb.checkpoint.transfer.thread.num: {threadNumberAccordingYourProjectSize}如果您不添加这个，它将是1(默认)
https://stackoverflow.com/questions/64158433
复制相似问题
腾讯云开发者
