。

然而#xff0c;当处理大型状态并频繁读写时#xff0c;可能会导致背压问题#xff0c;因为RocksDB需要从磁盘读取和写入数据#xff0c;而这可能成为瓶颈。

Flink开发…RocksDB是Flink中用于持久化状态的默认后端它提供了高性能和可靠的状态存储。

然而当处理大型状态并频繁读写时可能会导致背压问题因为RocksDB需要从磁盘读取和写入数据而这可能成为瓶颈。

webUI

从图上看瓶颈卡在序列化与反序列化结合业务逻辑代码业务涉及state大对象的读写并且是ValueState。

问题分析

V数据结构的状态可能会认为使用ValueStateHashMapK,

V或者使用MapStateK,

如果我们选择使用HashMap状态后端那么两种方式的性能上不会有很大差异但是如果我们选择使用RocksDB状态后端则推荐使用MapStateK,

V序列化为字节数组之后写入的。

同样在读取时也是先读取到字节数组然后反序列化为一整个HashMapK,

所以每次访问和更新ValueState时实际上都是对HashMapK,

V这个集合类的大对象做序列化以及反序列化如上图所示而这是一个及其耗费资源的过程很容易就会导致Flink作业产生性能瓶颈所以极不推荐在ValueState中存储大对象。

问题调优

使用MapState代替ValueState精简state数据量问题解决。

除此之外还有一些其他调优策略

增加算子并行度通过增加算子的并行度可以将负载分布到多个任务实例上从而提高整体处理能力。

这样可以减少单个任务实例的读写压力。

增加RocksDB的内存限制RocksDB使用内存来缓存热数据可以通过增加RocksDB的内存限制来提高缓存效果。

可以通过Flink的配置选项state.backend.rocksdb.memory.managed来设置内存限制。

调整RocksDB的配置参数RocksDB有许多配置参数可以调整以优化读写性能。

可以根据具体情况调整参数例如write_buffer_size、max_write_buffer_number、max_background_compactions等。

使用异步快照Flink提供了异步快照机制可以将状态异步地持久化到RocksDB。

这可以减少同步写入RocksDB的开销并提高整体吞吐量。

定期进行状态清理和压缩定期清理过期的状态数据和进行状态压缩可以减少磁盘占用和提高读写性能。

可以使用Flink的状态后端接口进行定期的清理和压缩操作等等。

拿当前实例来说虽然调优有很多种方式例如可以通过增量subtask并行度来提高处理能力数据量大了还是会出现处理上的瓶颈不可持续但并没有找出问题的根本原因或者说调优最大化。

通过使用MapState代替ValueState精简state数据量可以达到调优最大化以最小的代价解决最大的问题。