乐趣区

关于flink:Flink之状态管理State-Backends

jiezi

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 工作继续运行可能导致的状态数量暴增而内存不足的状况，因而适宜在生产环境应用。

退出移动版