Flink作业RocksDB状态后端调优

Flink作业RocksDB介绍

当启用RocksDB作为作业的状态后端时，大量的状态数据会导致RocksDB的读写性能差。可通过如下方法排查算子性能是否受RocksDB影响：

• 在TaskManager页面的ThreadDump查看算子是否长时间执行在RocksDB的操作接口上，多次刷新后出现如下所示即长时间执行在RocksDB的操作接口上。

  Join[5] -> Calc[6] -> Sink: print[7] (1/1)#0" Id=113 RUNNABLE (in native)
  	at org.rocksdb.RocksDB.put(Native Method)
  	at org.rocksdb.RocksDB.put(RocksDB.java:955)
  	at org.apache.flink.contrib.streaming.state.RocksDBValueState.update(RocksDBValueState.java:103)

• 通过开启火焰图（自定义配置rest.flamegraph.enabled=true打开火焰图）重新提交作业查看算子热点，如下图所示算子热点达到100%。
  图1 通过火焰图查看算子热点
  

开启Flink作业RocksDB监测步骤

1. 使用具有FlinkServer管理员权限的用户登录FusionInsight Manager。
2. 选择“集群 > 服务 > Flink”，在“Flink WebUI”右侧，单击链接，访问Flink的WebUI。
3. 单击“作业管理”进入作业管理页面。
4. 找到待调优的并处于非运行中的作业，在“操作”列单击“开发”进入作业开发界面。
5. 在作业开发界面的“自定义参数”项中，添加如下参数并保存。
   • 开启RocksDB监测

     表1 RocksDB监测配置

     参数名称	值	说明
     state.backend.rocksdb.metrics.hot.enabled	true	Rocksdb非统计的监测，非统计的监测包含Rocksdb Property包含的监测项
     state.backend.rocksdb.metrics.statistics.enabled	true	Rocksdb Statistics统计监测
     state.backend.rocksdb.metrics.num-immutable-mem-table	true	监测RocksDB中不可变Memtable的数量，该值如果一直增加，或大于设置的阈值，会影响写性能
     state.backend.rocksdb.metrics.mem-table-flush-pending	true	监测RocksDB中Pending的Memtable flush数
     state.backend.rocksdb.metrics.compaction-pending	true	监测RocksDB中Pending的Compaction数。如果Pending Compaction存在，则返回“1”，否则返回“0”
     state.backend.rocksdb.metrics.background-errors	true	监测RocksDB中metrics.background-errors的数量
     state.backend.rocksdb.metrics.cur-size-active-mem-table	true	监测Active Memtable的大致大小，单位：字节
     state.backend.rocksdb.metrics.cur-size-all-mem-tables	true	监测Active和未Flush的不可变Memtable的大致大小，单位：字节
     state.backend.rocksdb.metrics.size-all-mem-tables	true	监测Active memtable、未Flush的和Pinned的Memtable的大致大小，单位：字节
     state.backend.rocksdb.metrics.num-entries-active-mem-table	true	监测Active memtable中的条目总数
     state.backend.rocksdb.metrics.num-entries-imm-mem-tables	true	监测imm-mem-tables中的条目总数
     state.backend.rocksdb.metrics.num-deletes-active-mem-table	true	监测Active memtable中删除条目的总数
     state.backend.rocksdb.metrics.num-deletes-imm-mem-tables	true	监测未刷新的不可变Memtable中删除条目的总数
     state.backend.rocksdb.metrics.estimate-num-keys	true	监测RocksDB中的Key数量
     state.backend.rocksdb.metrics.estimate-table-readers-mem	true	监测用于读取SST表的内存，不包括块缓存（如过滤器和索引块）中使用的内存，单位：字节
     state.backend.rocksdb.metrics.num-snapshots	true	监测数据库未发布快照的数量
     state.backend.rocksdb.metrics.num-live-versions	true	监测实时版本的数量。版本是内部数据结构，版本太多说明RocksDB可能会因为查询或者Compaction而不能删除旧版本
     state.backend.rocksdb.metrics.estimate-live-data-size	true	监测实时数据量，单位：字节（由于空间放大，通常小于SST文件大小）
     state.backend.rocksdb.metrics.total-sst-files-size	true	监测所有版本的SST文件的总大小，单位：字节。文件太多，可能会降低查询的速度
     state.backend.rocksdb.metrics.live-sst-files-size	true	监测属于最新版本的所有SST文件的总大小，单位：字节。文件太多，可能会降低查询的速度
     state.backend.rocksdb.metrics.estimate-pending-compaction-bytes	true	监测Compaction的数据总大小，单位：字节。以使所有级别降至目标大小以下，基于级别以外的其他压缩无效
     state.backend.rocksdb.metrics.num-running-compactions	true	监测当前运行的Compaction数量，如果线程数都是Running，可能会影响写性能
     state.backend.rocksdb.metrics.num-running-flushes	true	监测当前运行的Flush任务数，如果线程数都是Running，可能会影响写性能
     state.backend.rocksdb.metrics.actual-delayed-write-rate	true	监测当前实际延迟写入速率。返回0表示无延迟
     state.backend.rocksdb.metrics.is-write-stopped	true	监测RocksDB中的写入是否已停止。如果写入已停止，则返回1，否则返回0
     state.backend.rocksdb.metrics.block-cache-capacity	true	监测Block cache容量
     state.backend.rocksdb.metrics.block-cache-usage	true	监测Block cache中数据占用内存大小
     state.backend.rocksdb.metrics.block-cache-pinned-usage	true	监测Block cache中pinned数据占用内存大小
     state.backend.rocksdb.metrics.compression-ratio	true	监测每层的压缩率
     state.backend.rocksdb.metrics.compression-ratio-levelN	7	监测压缩率的层数，取值范围：0-配置的层数
     state.backend.rocksdb.metrics.num-files	true	监测每层的文件数
     state.backend.rocksdb.metrics.num-files-levelN	7	监测文件数的层数，取值范围：0-配置的层数
     state.backend.rocksdb.metrics.statistics.ticker	block.cache.miss,block.cache.hit,block.cache.index.miss,block.cache.index.hit,block.cache.filter.miss,block.cache.filter.hit,block.cache.data.miss,block.cache.data.hit,bloom.filter.useful,memtable.hit,memtable.miss,l0.hit,l1.hit,l2andup.hit,stall.micros	statistics ticker监测项 如需添加新的监测项，在其末尾追加并以逗号分隔
     state.backend.rocksdb.metrics.statistics.histogram	db.get.micros,db.write.micros,db.flush.micros,compaction.times.micros	statistics直方监测项 如需添加新的监测项，在其末尾追加并以逗号分隔

   • 开启RocksDB告警

     表2 RocksDB告警配置

     配置项	默认值	说明
     metrics.reporter.alarm.job.alarm.rocksdb.metrics.enable	true	是否开启RocksDB监测，默认不开启。只有状态后端为RocksDB时该配置才有效
     metrics.reporter.alarm.job.alarm.rocksdb.metrics.duration	180s	RocksDB监测转告警的周期
     metrics.reporter.alarm.job.alarm.rocksdb.metrics.print.enabled	true	是否将RocksDB监测打印到TaskManager 当metrics.reporter.alarm.job.alarm.rocksdb.metrics.enable=true时，该配置项默认为true
     metrics.reporter.alarm.job.alarm.rocksdb.metrics.print.interval	5min	RocksDB监测打印到TaskManager的间隔时间
     metrics.reporter.alarm.job.alarm.rocksdb.get.micros.threshold	1000	Get耗时阈值，周期（metrics.reporter.alarm.job.alarm.rocksdb.metrics.duration）内连续出现超过该阈值，作业将上报告警，单位：微秒
     metrics.reporter.alarm.job.alarm.rocksdb.write.micros.threshold	3000	Write耗时阈值，周期（metrics.reporter.alarm.job.alarm.rocksdb.metrics.duration）内连续出现超过该阈值，作业将上报告警，单位：微秒
     metrics.reporter.alarm.job.alarm.actual-delayed-write-rate.threshold	0	周期（metrics.reporter.alarm.job.alarm.rocksdb.metrics.duration）内连续出现写限速，作业将上报告警，“0”表示不限速
     metrics.reporter.alarm.job.alarm.rocksdb.background.jobs.multiplier	2	flush/compaction的请求量超过了state.backend.rocksdb.thread.num的multiplier倍数，作业将上报告警

6. 在“作业管理”页面单击“启动”运行作业。然后根据RocksDB监测和告警情况，在作业开发界面的“自定义参数”项中添加如下参数调优作业。作业调优完成后建议关闭RocksDB的监测和告警。

   表3 RocksDB调优参数配置

   参数名称	值	说明
   state.backend.rocksdb.writebuffer.count	2	设置Active memtable和Immutable memtable数，当写入过快或者Flink线程太少，会导致写入阻塞。启用SPINNING_DISK_OPTIMIZED_HIGH_MEM时，默认值为“4” 建议该值大于等于“state.backend.rocksdb.writebuffer.number-to-merge”的值加“2”
   state.backend.rocksdb.writebuffer.size	64MB	Memtable大小
   state.backend.rocksdb.thread.num	2	RocksDB Flush和Compaction的线程数，启用SPINNING_DISK_OPTIMIZED_HIGH_MEM时，默认值为“4”
   state.backend.rocksdb.writebuffer.number-to-merge	1	Immutable flush前的个数，n个Immutable flush时会去重，启用SPINNING_DISK_OPTIMIZED_HIGH_MEM时，默认值是“3”
   state.backend.rocksdb.compaction.level.max-size-level-base	256MB	Level1的SSL文件总大小，启用SPINNING_DISK_OPTIMIZED_HIGH_MEM时，默认值是“1GB”
   state.backend.rocksdb.compaction.level.target-file-size-base	64MB	Level1+的SSL文件大小，启用SPINNING_DISK_OPTIMIZED_HIGH_MEM时，默认值是“128MB”
   state.backend.rocksdb.num_levels	7	RocksDB的Level数
   state.backend.rocksdb.level0_slowdown_writes_trigger	20	Level0触发Slowdown的文件数，小于“0”表示永远不会触发
   state.backend.rocksdb.level0_stop_writes_trigger	36	Level0触发Stop的最大文件数
   state.backend.rocksdb.max_compaction_bytes	-	一次Compaction最大的Bytes，默认值：（state.backend.rocksdb.compaction.level.target-file-size-base） * 25
   state.backend.rocksdb.level0_file_num_compaction_trigger	4	Level0的SST数量达到阈值，触发Level0到Level1的Compaction
   state.backend.rocksdb.compression	snappy	SST文件压缩算法 取值范围：null、snapp、zlib、bzip2、lz4、lz4hc、xpress、zstd
   state.backend.rocksdb.bottommost_compression	snappy	底层使用重量级的压缩类型，减少空间。因为底层的数据可能是冷数据，如果要启用，推荐使用zstd或者zlib 取值范围：null、snapp、zlib、bzip2、lz4、lz4hc、xpress、zstd
   state.backend.rocksdb.max_bytes_for_level_multiplier	10	Level1加相邻2层的数据量倍数因子
   state.backend.rocksdb.hard-pending-compaction-bytes-limit	256GB	当Pending的Compaction超过该阈值，写停止
   state.backend.rocksdb.soft-pending-compaction-bytes-limit	64GB	当Pending的Compaction超过该阈值，写限流
   state.backend.rocksdb.use-bloom-filter	true	Bloom过滤器，开启后每个新创建的SST文件都将包含一个Bloom过滤器
   state.backend.rocksdb.block.cache-size	8MB	Cache缓存大小，启用SPINNING_DISK_OPTIMIZED_HIGH_MEM时，默认值是“256MB”
   state.backend.rocksdb.block.blocksize	4KB	Block大小，启用SPINNING_DISK_OPTIMIZED_HIGH_MEM时，默认值是“128KB”
   state.backend.rocksdb.files.open	-1	最大打开句柄数，主要被用在SST文件句柄上，“-1”表示不限制