Flink提供了以下三种开箱即用的状态后端(用于存储状态数据)
- MemoryStateBackend
- FsStateBackend
- RocksDBStateBackend
MemoryStateBackend
MemoryStateBackend外部将state作为对象保留在taskManager的堆内存中，通过checkpoint机制，MemoryStateBackend将state进行快照并保留Jobmanager的堆内存中。
MemoryStateBackend能够通过配置来应用异步快照（asynchronous snapshots），通过异步快照能够防止阻塞管道，目前是默认开启。
MemoryStateBackend的限度：
- 每个独立的状态（state）默认限度大小为5MB， 能够通过构造函数减少容量；
- 状态的大小不能超过akka的framesize大小。
- 聚合状态（aggregate state ）必须放入JobManager的内存。
MemoryStateBackend的实用场景：
- 本地调试
- flink工作状态数据量较小的场景
FsStateBackend
FsStateBackend通过配置文件系统门路来进行设置，将动态数据保留在taskmanger的内存中，通过checkpoint机制，将状态快照写入配置好的文件系统或目录中。
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//fs状态后端配置,如为file:///,则在taskmanager的本地val checkPointPath = new Path("hdfs:///flink/checkpoints") val fsStateBackend: StateBackend = new FsStateBackend(checkPointPath)env.setStateBackend(fsStateBackend)FsStateBackend实用场景：
- 大状态、长窗口、大key/value状态的的工作
- 全高可用配置
RocksDBStateBackend
RocksDBStateBackend将工作状态保留在RocksDB数据库（地位在taskManagerd的数据目录）。通过checkpoint, 整个RocksDB数据库被复制到配置的文件系统或目录中
private val checkpointDataUri = "hdfs:///flink/checkpoints" private val tmpDir = "file:///tmp/rocksdb/data/" val env = StreamExecutionEnvironment.getExecutionEnvironment val fsStateBackend: StateBackend = new FsStateBackend(checkpointDataUri) val rocksDBBackend: RocksDBStateBackend = new RocksDBStateBackend(fsStateBackend, TernaryBoolean.TRUE) val config = new Configuration() //TIMER分为HEAP(默认，性能更好)和RocksDB(扩大好) config.setString(RocksDBOptions.TIMER_SERVICE_FACTORY,RocksDBStateBackend.PriorityQueueStateType.ROCKSDB.toString) rocksDBBackend.configure(config) rocksDBBackend.setDbStoragePath(tmpDir) env.setStateBackend(rocksDBBackend.asInstanceOf[StateBackend])RocksDBStateBackend实用场景：
- 大状态、长窗口、大key/value状态的的工作
- 全高可用配置
因为RocksDBStateBackend将工作状态存储在taskManger的本地文件系统，状态数量仅仅受限于本地磁盘容量限度，比照于FsStateBackend保留工作状态在内存中，RocksDBStateBackend能防止flink工作继续运行可能导致的状态数量暴增而内存不足的状况，因而适宜在生产环境应用。