Flink中如何为特定Keyed State配置RocksDB StateBackend参数

核心实现思路 #

Flink 的 RocksDBStateBackend 支持为不同状态（底层对应独立的 RocksDB 列族）配置专属的列族参数，你可以通过自定义列族选项工厂，给小体量的 MapState 分配独立的块缓存保障读性能，大体量状态走全局托管内存避免总内存超配，同时针对两种状态的读写特性分别做优化。

具体实现步骤 #

• 第一步：为不同状态定义唯一标识的名称
  在 ProcessFunction 的 open 方法中初始化状态时，给两个 MapState 配置不同的状态名，方便后续识别匹配：

public static class MyProcessFunction1 extends KeyedProcessFunction<Integer, String, Long> {
    private transient MapState<byte[], byte[]> largeMapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor<byte[], byte[]> largeStateDesc = new MapStateDescriptor<>(
            "large_map_state", // 大状态唯一名称
            byte[].class,
            byte[].class
        );
        largeMapState = getRuntimeContext().getMapState(largeStateDesc);
    }
}

public static class MyProcessFunction2 extends KeyedProcessFunction<Integer, String, Long> {
    private transient MapState<byte[], byte[]> smallMapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor<byte[], byte[]> smallStateDesc = new MapStateDescriptor<>(
            "small_map_state", // 小状态唯一名称
            byte[].class,
            byte[].class
        );
        smallMapState = getRuntimeContext().getMapState(smallStateDesc);
    }
}

• 第二步：实现自定义 RocksDB 选项工厂，差异化配置列族参数

import org.rocksdb.*;
import org.apache.flink.contrib.streaming.state.RocksDBOptionsFactory;
import org.apache.flink.contrib.streaming.state.ColumnFamilyOptionsConfig;
import java.util.Collection;

public class CustomStateOptionsFactory implements RocksDBOptionsFactory {
    // 给小状态单独分配512MB专属LRU块缓存，全局单例复用，不占用托管内存的共享缓存
    private static final Cache SMALL_STATE_SPECIAL_CACHE = new LRUCache(512 * 1024 * 1024);

    @Override
    public DBOptions createDBOptions(DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
        // 全局DB级配置保持默认即可，有全局优化需求可在此调整
        return currentOptions;
    }

    @Override
    public ColumnFamilyOptions createColumnFamilyOptions(ColumnFamilyOptions currentOptions, ColumnFamilyOptionsConfig config, Collection<AutoCloseable> handlesToClose) {
        String stateName = config.getStateDescriptor().getName();
        if ("small_map_state".equals(stateName)) {
            // 小状态读性能优化配置
            BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
            tableConfig.setBlockCache(SMALL_STATE_SPECIAL_CACHE);
            tableConfig.setBlockSize(4 * 1024); // 小KV用更小的块提升缓存命中率
            tableConfig.setCacheIndexAndFilterBlocks(true); // 索引和过滤块也存入缓存，进一步降低读延迟
            currentOptions.setTableFormatConfig(tableConfig);
            // 额外针对小型DB优化
            currentOptions.setCompressionType(CompressionType.NO_COMPRESSION); // 小数据不需要压缩，节省CPU
            currentOptions.setLevel0FileNumCompactionTrigger(100); // 拉高compaction触发阈值，减少后台IO
        } else if ("large_map_state".equals(stateName)) {
            // 大状态写性能优化配置
            BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
            tableConfig.setBlockCache(null); // 禁用专属缓存，走全局托管内存的共享缓存
            tableConfig.setBlockSize(32 * 1024); // 大KV用更大的块减少元数据开销
            currentOptions.setTableFormatConfig(tableConfig);
            // 写优化配置
            currentOptions.setCompressionType(CompressionType.LZ4_COMPRESSION); // 开启轻量压缩减少磁盘占用
            currentOptions.setWriteBufferSize(128 * 1024 * 1024); // 增大写缓冲区减少刷盘次数
            currentOptions.setMaxWriteBufferNumber(4);
        }
        // 其他未匹配状态走默认配置
        return currentOptions;
    }
}

• 第三步：注册自定义选项工厂到状态后端
  在作业启动代码中，给 RocksDBStateBackend 绑定你实现的自定义工厂即可：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 初始化RocksDB状态后端
RocksDBStateBackend rocksDBBackend = new RocksDBStateBackend("你的checkpoint存储路径");
// 绑定自定义选项工厂
rocksDBBackend.setRocksDBOptions(new CustomStateOptionsFactory());
env.setStateBackend(rocksDBBackend);

注意事项 #

• 小状态的专属块缓存属于Flink托管内存之外的堆外直接内存，你需要在作业启动参数中调大 taskmanager.memory.task.off-heap.size，预留足够的内存给专属缓存，避免出现直接内存OOM
• 专属缓存大小建议设置为实际数据量的1.3倍以上，保证全量小状态数据都能落在缓存中，实现最高读QPS
• 如果使用Flink 1.13以下版本，无法直接通过config.getStateDescriptor().getName()获取状态名，可以解析config.getColumnFamilyName()的前缀，列族命名规则为{状态名}-{分区后缀}，截取前缀即可匹配状态

内容的提问来源于stack exchange，提问作者Shenjiaqi