文章引自：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)來進行設定，例如：“

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配置選項：

2. 單任務配置
   通過在單個flink任務中通過env.setStateBackend(...)單獨調整state backend配置，這種方式會覆蓋全域性配置。例如：

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

四、總結

本文介紹了flink狀態後端的三種配置方式和區別，並介紹了狀態後端的配置方法。在生產環境中，對於大狀態量應用，推薦使用RocksDBStateBackend進行狀態後端配置，以應對可能存在的記憶體不足情況。