RocksDB状态后端LZ4压缩对Flink-SQL应用的影响及技术问询

背景信息 #

我正在使用Flink 1.17.1运行Flink-SQL应用，示例SQL代码如下：

create table enrich_table (
col1 STRING,
col2....
)
WITH ( 'connector' = 'upsert-kafka',
    'topic' = 'enrich_table',
    'properties.bootstrap.servers' = '${KAFKA_BROKERS}',
    'properties.group.id' = '${CONSUMER_GROUP_ID}',
    'value.format' = 'avro-confluent',
    'value.avro-confluent.url' = '${SCHEMA_REGISTRY_URL}',
    'key.format' = 'avro-confluent',
    'key.avro-confluent.url' = '${SCHEMA_REGISTRY_URL}');


Insert into enrich_table as 
select
    col1,col2...

from 
    table1 t1 
    inner join table2 t2 on ..
    inner join table3 t2 on ...
    .....

通过自定义工厂类为RocksDB设置LZ4压缩，Java代码如下：

package my.org.abc

import org.rocksdb.{ColumnFamilyOptions, DBOptions}
import org.apache.flink.configuration.ReadableConfig
import org.apache.flink.contrib.streaming.state.{ConfigurableRocksDBOptionsFactory, DefaultConfigurableOptionsFactory, RocksDBOptionsFactory}


import org.apache.flink.configuration.ConfigOptions
import org.apache.flink.configuration.ConfigOption

import org.apache.flink.configuration.description.LinkElement.link
import org.rocksdb.CompressionType
import org.apache.flink.configuration.description.Description


class RocksDBTuningJobOptionsFactory extends ConfigurableRocksDBOptionsFactory {
  private val defaultFactory = new DefaultConfigurableOptionsFactory
  private var configuration: ReadableConfig = _
  private val COMPRESSION: ConfigOption[CompressionType] =
    ConfigOptions.key("state.backend.rocksdb.custom.compression")
      .enumType(classOf[CompressionType])
      .noDefaultValue()
      .withDescription(
        Description.builder()
          .text("Configures RocksDB compression")
          .linebreak()
          .text(
            "For more information, please refer to %s",
            link("https://github.com/facebook/rocksdb/wiki/Compression")
          )
          .build()
      )

  override def configure(configuration: ReadableConfig): RocksDBOptionsFactory = {
    defaultFactory.configure(configuration)
    this.configuration = configuration
    this
  }

  override def createDBOptions(currentOptions: DBOptions, handlesToClose: java.util.Collection[AutoCloseable]): DBOptions = {
    defaultFactory.createDBOptions(currentOptions, handlesToClose)
  }

  override def createColumnOptions(currentOptions: ColumnFamilyOptions, handlesToClose: java.util.Collection[AutoCloseable]): ColumnFamilyOptions = {
    var updatedOptions = defaultFactory.createColumnOptions(currentOptions, handlesToClose)

    configuration.getOptional(COMPRESSION).ifPresent(currentOptions.setCompressionType)
    updatedOptions
  }
}

配置参数如下：

........
state.backend.rocksdb.use-bloom-filter: "true"
state.backend.rocksdb.options-factory: 'com.mdsol.streaming.util.RocksDBTuningJobOptionsFactory'
state.backend.rocksdb.custom.compression: "LZ4_COMPRESSION"

问题解答 #

1. 大状态、高吞吐高流速且无时态Join的Flink-SQL应用中，社区常用的压缩方式是什么？ #

在这类场景下，LZ4是社区最常用的选择之一——它压缩解压速度极快，CPU开销极低，能在大幅减少状态存储空间的同时，几乎不影响高吞吐场景下的性能表现。如果对压缩率有更高要求，且能接受稍高一点的CPU开销，ZSTD也是不少团队的备选，它的压缩比比LZ4更优，且在新版本RocksDB中性能优化得不错。而Snappy、GZIP这类要么压缩率不足，要么速度太慢，在高吞吐大状态场景下应用较少。

2. 设置LZ4压缩是否必须继承ConfigurableRocksDBOptionsFactory，有无更优实现方式？ #

完全不需要继承这个类。更简洁高效的方式是直接用Flink内置配置参数：从Flink 1.10版本开始，就支持通过state.backend.rocksdb.compression.type参数直接指定压缩类型，比如设置为LZ4就能启用LZ4压缩，完全不用写自定义代码。

只有当你需要更灵活的配置（比如给不同ColumnFamily设置不同压缩策略）时，才需要考虑自定义OptionsFactory。单纯全局设置LZ4压缩的话，内置参数是最优方案，既简洁又不容易出错。

3. 对已运行且存在Checkpoint的作业修改为LZ4压缩会产生哪些影响？ #

• 首次恢复耗时增加：从旧压缩方式（比如默认无压缩或其他压缩格式）切换到LZ4后，作业从Checkpoint恢复时，RocksDB会自动将旧状态数据转换为LZ4格式，状态越大，转换耗时越长。
• 运行时性能提升：恢复完成后，后续状态读写都会采用LZ4压缩，存储空间减少，磁盘IO压力降低，高吞吐场景下性能会有明显提升，尤其对原本存在磁盘IO瓶颈的作业效果更显著。
• Checkpoint体积缩小：新生成的Checkpoint会用LZ4压缩，体积比之前更小，节省持久化存储资源，同时Checkpoint的写入时间也会缩短。
• 兼容性无风险：同一Flink版本内切换压缩方式不会有兼容性问题，RocksDB能兼容不同压缩格式的状态转换；但如果后续切回旧压缩方式，同样会经历一次状态转换过程。

内容的提问来源于stack exchange，提问作者Faisal Ahmed Siddiqui