flink的状态后端,以及RocksDB StateBackend的配置

文章引自:https://www.meiwen.com.cn/subject/ypmbpctx.html

 

  flink提供不同的状态后端(state backends)来区分状态的存储方式和存储位置。flink状态可以存储在java堆内存内或者内存之外。通过状态后端的设置,flink允许应用保持大容量的状态。开发者可以在不改变应用逻辑的情况下设置状态后端。
         默认情况下,flink的状态会保存在taskmanager的内存中,而checkpoint会保存在jobManager的内存中。

二、可用的State Backend

flink提供三种开箱即用的State Backend:

  • MemoryStateBackend
  • FsStateBackend
  • RocksDBStateBackend

如果没有配置,则默认使用MemoryStateBackend。

2.1 MemoryStateBackend

        MemoryStateBackend内部将状态(state)数据作为对象保存在java堆内存中(taskManager),通过checkpoint机制,MemoryStateBackend将状态(state)进行快照并保存Jobmanager(master)的堆内存中。

        MemoryStateBackend可以通过配置来使用异步快照(asynchronous snapshots)。通过异步快照可以避免阻塞管道(blocking pipelines),目前是默认开启,当然也可以通过MemoryStateBackend的构造函数配置进行关闭:

new MemoryStateBackend(MAX_MEM_STATE_SIZE, false);

MemoryStateBackend的限制:

  • 每个独立的状态(state)默认限制大小为5MB, 可以通过构造函数增加容量;
  • 状态的大小不能超过akka的framesize大小。参考:配置 ;
  • 聚合状态(aggregate state )必须放入JobManager的内存。

MemoryStateBackend的适用场景:

  • 本地调试
  • flink任务状态数据量较小的场景

2.2 FsStateBackend

        FsStateBackend通过配置文件系统路径(type, address, path)来进行设置,例如:“hdfs://namenode:40010/flink/checkpoints” 或者 “file:///data/flink/checkpoints”.
        FsStateBackend将动态数据保存在taskmanger的内存中,通过checkpoint机制,将状态快照写入配置好的文件系统或目录中。最小元数据保存jobManager的内存中,另外FsStateBackend通过配置一个fileStateThreshold阈值,小于该值时state存储到metadata中而非文件中。

         FsStateBackend默认通过配置来使用异步快照(asynchronous snapshots)避免阻塞管道(blocking pipelines),当然也可以通过FsStateBackend的构造函数配置进行关闭:

new FsStateBackend(path, false);

FsStateBackend适用场景:

  • 大状态、长窗口、大key/value状态的的任务
  • 全高可用配置

2.3 RocksDBStateBackend

如果使用java代码进行单任务配置,使用前先加入依赖:

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
            <version>1.8.0</version>
        </dependency>

        RocksDBStateBackend也通过配置文件系统路径来配置,例如:“hdfs://namenode:40010/flink/checkpoints” 或者 “file:///data/flink/checkpoints”,RocksDBStateBackend内部默认会构造一个FsStateBackend来初始化checkpointStreamBackend属性,在实现StateBackend接口的resolveCheckpoint和createCheckpointStorage方法时实际上调用checkpointStreamBackend的相关方法来实现:

    @Override
    public CompletedCheckpointStorageLocation resolveCheckpoint(String pointer) throws IOException {
        return checkpointStreamBackend.resolveCheckpoint(pointer);
    }

    @Override
    public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException {
        return checkpointStreamBackend.createCheckpointStorage(jobId);
    }

        RocksDBStateBackend将工作状态保存在RocksDB数据库(位置在taskManagerd的数据目录)。通过checkpoint, 整个RocksDB数据库被复制到配置的文件系统或目录中。最小元数据保存jobManager的内存中。RocksDBStateBackend可以通过enableIncrementalCheckpointing参数配置是否进行增量Checkpoint(而MemoryStateBackend 和 FsStateBackend不能)。

        跟FsStateBackend 不同的是,RocksDBStateBackend仅支持异步快照(asynchronous snapshots)。

RocksDBStateBackend适用场景:

  • 大状态、长窗口、大key/value状态的的任务
  • 全高可用配置
    由于RocksDBStateBackend将工作状态存储在taskManger的本地文件系统,状态数量仅仅受限于本地磁盘容量限制,对比于FsStateBackend保存工作状态在内存中,RocksDBStateBackend能避免flink任务持续运行可能导致的状态数量暴增而内存不足的情况,因此适合在生产环境使用。

三、配置方式

  1. 全局配置
    flink可以通过flink-conf.yaml 配置原因全局配置state backend。
    使用 state.backend 选项进行state backend类型配置:可选值包括: jobmanager (MemoryStateBackend), filesystem (FsStateBackend), rocksdb (RocksDBStateBackend)。
    使用state.checkpoints.dir选项设置checkpoints数据和元数据文件。

一个简单的配置形式如下:

# The backend that will be used to store operator state checkpoints
state.backend: filesystem
# Directory for storing checkpoints
state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints

RocksDBStateBackend配置选项:

Key Default Description
state.backend.rocksdb.checkpoint.transfer.thread.num 1 The number of threads used to transfer (download and upload) files in RocksDBStateBackend.
state.backend.rocksdb.localdir (none) The local directory (on the TaskManager) where RocksDB puts its files.
state.backend.rocksdb.options-factory "org.apache.flink.contrib.streaming.state.DefaultConfigurableOptionsFactory" The options factory class for RocksDB to create DBOptions and ColumnFamilyOptions. The default options factory is org.apache.flink.contrib.streaming.state.DefaultConfigurableOptionsFactory, and it would read the configured options which provided in 'RocksDBConfigurableOptions'.
state.backend.rocksdb.predefined-options "DEFAULT" The predefined settings for RocksDB DBOptions and ColumnFamilyOptions by Flink community. Current supported candidate predefined-options are DEFAULT, SPINNING_DISK_OPTIMIZED, SPINNING_DISK_OPTIMIZED_HIGH_MEM or FLASH_SSD_OPTIMIZED. Note that user customized options and options from the OptionsFactory are applied on top of these predefined ones.
state.backend.rocksdb.timer-service.factory "HEAP" This determines the factory for timer service state implementation. Options are either HEAP (heap-based, default) or ROCKSDB for an implementation based on RocksDB .
state.backend.rocksdb.ttl.compaction.filter.enabled false This determines if compaction filter to cleanup state with TTL is enabled for backend.Note: User can still decide in state TTL configuration in state descriptor whether the filter is active for particular state or not.
  1. 单任务配置
            通过在单个flink任务中通过env.setStateBackend(...)单独调整state backend配置,这种方式会覆盖全局配置。例如:
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints"));

四、总结

        本文介绍了flink状态后端的三种配置方式和区别,并介绍了状态后端的配置方法。在生产环境中,对于大状态量应用,推荐使用RocksDBStateBackend进行状态后端配置,以应对可能存在的内存不足情况。

上一篇:Flink State 和 Fault Tolerance


下一篇:入门大数据---Flink状态管理与检查点机制