分类
大数据

Flink的状态后端(State Backends)

当使用checkpoint时,状态(state)会被持久化到checkpoint上,以防止数据的丢失并确保发生故障时能够完全恢复。状态是通过什么方式在哪里持久化,取决于使用的状态后端。

可用的状态后端

  • MemoryStateBackend
  • FsStateBackend
  • RocksDBStateBackend

注意:如果什么都不配置,系统默认的是MemoryStateBackend

MemoryStateBackend

MemoryStateBackend 是将状态维护在 Java 堆上的一个内部状态后端。键值状态和窗口算子使用哈希表来存储数据(values)和定时器(timers)。当应用程序 checkpoint 时,此后端会在将状态发给 JobManager 之前快照下状态,JobManager 也将状态存储在 Java 堆上。默认情况下,MemoryStateBackend 配置成支持异步快照。异步快照可以避免阻塞数据流的处理,从而避免反压的发生。当然,使用 new MemoryStateBackend(MAX_MEM_STATE_SIZE, false)也可以禁用该特点。缺点:

  • 默认情况下,每一个状态的大小限制为 5 MB。可以通过 MemoryStateBackend 的构造函数增加这个大小。状态大小受到 akka 帧大小的限制(maxStateSize <= akka.framesize 默认 10 M),所以无论怎么调整状态大小配置,都不能大于 akka 的帧大小。也可以通过 akka.framesize 调整 akka 帧大小。
  • 状态的总大小不能超过 JobManager 的内存。

推荐使用的场景:

  • 本地测试、几乎无状态的作业,比如 ETL、JobManager 不容易挂,或挂掉影响不大的情况。
  • 不推荐在生产场景使用。

FsStateBackend

FsStateBackend需要配置的主要是文件系统,如 URL(类型,地址,路径)。比如可以是:“hdfs://namenode:40010/flink/checkpoints” 或“s3://flink/checkpoints”当选择使用 FsStateBackend时,正在进行的数据会被存在TaskManager的内存中。在checkpoint时,此后端会将状态快照写入配置的文件系统和目录的文件中,同时会在JobManager的内存中(在高可用场景下会存在 Zookeeper 中)存储极少的元数据。容量限制上,单 TaskManager 上 State 总量不超过它的内存,总大小不超过配置的文件系统容量。默认情况下,FsStateBackend 配置成提供异步快照,以避免在状态 checkpoint 时阻塞数据流的处理。该特性可以实例化 FsStateBackend 时传入false的布尔标志来禁用掉,例如:new FsStateBackend(path, false)推荐使用的场景:

  • 处理大状态,长窗口,或大键值状态的有状态处理任务, 例如分钟级窗口聚合或 join。
  • 适合用于高可用方案(需要开启HA的作业)。
  • 可以在生产环境中使用

RocksDBStateBackend

RocksDBStateBackend 的配置也需要一个文件系统(类型,地址,路径),如下所示:“hdfs://namenode:40010/flink/checkpoints” 或“s3://flink/checkpoints”RocksDB 是一种嵌入式的本地数据库。RocksDBStateBackend 将处理中的数据使用 RocksDB 存储在本地磁盘上。在 checkpoint 时,整个 RocksDB 数据库会被存储到配置的文件系统中,或者在超大状态作业时可以将增量的数据存储到配置的文件系统中。同时 Flink 会将极少的元数据存储在 JobManager 的内存中,或者在 Zookeeper 中(对于高可用的情况)。RocksDB 默认也是配置成异步快照的模式。RocksDB是一个 key/value 的内存存储系统,和其他的 key/value 一样,先将状态放到内存中,如果内存快满时,则写入到磁盘中,但需要注意RocksDB不支持同步的 Checkpoint,构造方法中没有同步快照这个选项。不过RocksDB支持增量的 Checkpoint,也是目前唯一增量 Checkpoint 的 Backend,意味着并不需要把所有 sst 文件上传到 Checkpoint 目录,仅需要上传新生成的 sst 文件即可。它的 Checkpoint 存储在外部文件系统(本地或HDFS),其容量限制只要单个 TaskManager 上 State 总量不超过它的内存+磁盘,单Key最大2G,总大小不超过配置的文件系统容量即可。缺点:

  • RocksDB支持的单key和单value的大小最大为每个 2^31 字节。这是因为 RocksDB 的 JNI API 是基于byte[]的。
  • 对于使用具有合并操作的状态的应用程序,例如 ListState,随着时间可能会累积到超过 2^31 字节大小,这将会导致在接下来的查询中失败。

推荐使用的场景:

  • 最适合用于处理大状态,长窗口,或大键值状态的有状态处理任务。
  • 非常适合用于高可用方案。
  • 最好是对状态读写性能要求不高的作业

总结

那如何选择状态的类型和存储方式?结合前面的内容,可以看到,首先是要分析清楚业务场景;比如想要做什么,状态到底大不大。比较各个方案的利弊,选择根据需求合适的状态类型和存储方式即可。

原文:https://zhuanlan.zhihu.com/p/165906344

发表评论

电子邮件地址不会被公开。