【flink】flink狀態(tài)后端配置-設置State Backend

一忽肛、前言

????????flink提供不同的狀態(tài)后端（state backends）來區(qū)分狀態(tài)的存儲方式和存儲位置撬讽。flink狀態(tài)可以存儲在java堆內存內或者內存之外诉植。通過狀態(tài)后端的設置疫稿，flink允許應用保持大容量的狀態(tài)鸵隧。開發(fā)者可以在不改變應用邏輯的情況下設置狀態(tài)后端张弛。
???????? 默認情況下荒典，flink的狀態(tài)會保存在taskmanager的內存中宗挥，而checkpoint會保存在jobManager的內存中。

二种蝶、可用的State Backend

flink提供三種開箱即用的State Backend：

MemoryStateBackend
FsStateBackend
RocksDBStateBackend

如果沒有配置契耿，則默認使用MemoryStateBackend。

2.1 MemoryStateBackend

????????MemoryStateBackend內部將狀態(tài)（state）數據作為對象保存在java堆內存中（taskManager）螃征，通過checkpoint機制搪桂，MemoryStateBackend將狀態(tài)（state）進行快照并保存Jobmanager（master）的堆內存中。

????????MemoryStateBackend可以通過配置來使用異步快照（asynchronous snapshots）盯滚。通過異步快照可以避免阻塞管道（blocking pipelines）踢械，目前是默認開啟，當然也可以通過MemoryStateBackend的構造函數配置進行關閉：

new MemoryStateBackend(MAX_MEM_STATE_SIZE, false);

MemoryStateBackend的限制：

每個獨立的狀態(tài)（state）默認限制大小為5MB魄藕，可以通過構造函數增加容量内列；
狀態(tài)的大小不能超過akka的framesize大小。參考：配置背率；
聚合狀態(tài)（aggregate state ）必須放入JobManager的內存话瞧。

MemoryStateBackend的適用場景：

本地調試
flink任務狀態(tài)數據量較小的場景

2.2 FsStateBackend

????????FsStateBackend通過配置文件系統(tǒng)路徑(type, address, path)來進行設置，例如：“hdfs://namenode:40010/flink/checkpoints” 或者 “file:///data/flink/checkpoints”.
????????FsStateBackend將動態(tài)數據保存在taskmanger的內存中寝姿，通過checkpoint機制交排，將狀態(tài)快照寫入配置好的文件系統(tǒng)或目錄中。最小元數據保存jobManager的內存中饵筑，另外FsStateBackend通過配置一個fileStateThreshold閾值埃篓，小于該值時state存儲到metadata中而非文件中。

???????? FsStateBackend默認通過配置來使用異步快照（asynchronous snapshots）避免阻塞管道（blocking pipelines）根资，當然也可以通過FsStateBackend的構造函數配置進行關閉：

new FsStateBackend(path, false);

FsStateBackend適用場景：

大狀態(tài)架专、長窗口、大key/value狀態(tài)的的任務
全高可用配置

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也通過配置文件系統(tǒng)路徑來配置部脚，例如：“hdfs://namenode:40010/flink/checkpoints” 或者 “file:///data/flink/checkpoints”，RocksDBStateBackend內部默認會構造一個FsStateBackend來初始化checkpointStreamBackend屬性桨仿，在實現(xiàn)StateBackend接口的resolveCheckpoint和createCheckpointStorage方法時實際上調用checkpointStreamBackend的相關方法來實現(xiàn)：

    @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將工作狀態(tài)保存在RocksDB數據庫（位置在taskManagerd的數據目錄）睛低。通過checkpoint, 整個RocksDB數據庫被復制到配置的文件系統(tǒng)或目錄中。最小元數據保存jobManager的內存中服傍。RocksDBStateBackend可以通過enableIncrementalCheckpointing參數配置是否進行增量Checkpoint（而MemoryStateBackend 和 FsStateBackend不能）。

????????跟FsStateBackend 不同的是骂铁，RocksDBStateBackend僅支持異步快照（asynchronous snapshots）吹零。

RocksDBStateBackend適用場景：

大狀態(tài)、長窗口拉庵、大key/value狀態(tài)的的任務
全高可用配置
由于RocksDBStateBackend將工作狀態(tài)存儲在taskManger的本地文件系統(tǒng)灿椅，狀態(tài)數量僅僅受限于本地磁盤容量限制，對比于FsStateBackend保存工作狀態(tài)在內存中，RocksDBStateBackend能避免flink任務持續(xù)運行可能導致的狀態(tài)數量暴增而內存不足的情況茫蛹，因此適合在生產環(huán)境使用操刀。

三、配置方式

全局配置
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.

單任務配置
????????通過在單個flink任務中通過env.setStateBackend(...)單獨調整state backend配置柬采，這種方式會覆蓋全局配置欢唾。例如：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints"));

四、總結

????????本文介紹了flink狀態(tài)后端的三種配置方式和區(qū)別粉捻，并介紹了狀態(tài)后端的配置方法礁遣。在生產環(huán)境中，對于大狀態(tài)量應用肩刃，推薦使用RocksDBStateBackend進行狀態(tài)后端配置祟霍，以應對可能存在的內存不足情況。