Flink 的状态数据到底存到哪了？

之前看到群里好多同学提问：

• Flink 状态存在 heap 中，作业失败下次重启从 State 恢复，是不是数据就丢了？
• TM 本地状态恢复跟 Checkpoint 之间有什么关系？
• Operator State 是不是只能存在 heap 中？
  
  

如果有类似的疑问，相信本文能让你彻底搞懂 Flink 的状态数据到底存到了哪里。想加群的小伙伴可以在公众号菜单栏右下角联系博主。

0、结论

按照状态类型进行划分，State 分为 Operator State 和 KeyedState。

无论何种 StateBackend，Operator State 在运行期间全部存储在 TM 的内存中。Checkpoint 时，MemoryStateBackend 会将状态数据保存到 JM 的内存中。其他 StateBackend 会将状态数据保存到 hdfs。对应表格：

KeyedState 的存储表格如下所示：

如果对表格内容有一些疑惑，可以详细阅读下文。

大多数场景使用的都是 KeyedState，所以先分析 KeyedState 中的数据存储。

1、 keyedState 数据存储

1.1 KeyedState 的类型

众所周知：Flink 支持三种 StateBackend，分别是：

• memory 模式：MemoryStateBackend
• filesystem 模式：FsStateBackend
• rocksdb 模式：RocksDBStateBackend

不要被名字所迷惑，之前看到有同学在群里提问：filesystem 模式（FsStateBackend）会将状态数据存储在 hdfs 上，那读写性能会不会很差呢？是不是线上必须用 memory 模式或 rocksdb 模式呢？如果有这样的疑惑，显然是对 StateBackend 不了解。

上述问题的答案是这样的：生产环境不能使用 memory 模式，只能使用 filesystem 和 rocksdb 模式，且 filesystem 模式性能极高。

1.2 StateBackend 的作用

StateBackend 在 Flink 中的作用是存储状态数据。

例如：计算 pv 的场景，无论当前的 pv 值是 100 还是 500，都需要存储起来。pv 值将会存储在 StateBackend 中。

例如：计算 uv 的场景需要按照 userId 进行去重，必然要将所有的 userId 存储起来。这些 userId 就会存储在 StateBackend 中。

1.3 不同的 StateBackend 是如何工作的？

1.3.1 小状态场景，如何提供高性能的状态存储？

有些场景，状态数据特别小，所以 StateBackend 将状态数据放在内存中是个不错的选择。小数据量仅占用很少的内存空间，大部分任务都可以提供这些内存。

memory 模式（MemoryStateBackend）和 filesystem 模式（FsStateBackend）就是用到了这种思路，将状态数据存储在 TM 的内存中，而且使用类似于 HashMap 的 hash 结构。无论是查询还是写入的时间复杂度都可以达到 O(1)，从而提供了非常高的性能。

1.3.2 大状态场景，如何提供高性能的状态存储？

如果状态数据比较大，全放到内存中就会占用大量的内存，需要大量的经济成本。而且当状态数据比较大时，可能有一部分数据是热数据，一部分数据是冷数据，冷数据访问频率特别低也没必要全部放到内存中。所以大状态场景，数据全部存储在内存中可能不切合实际。

简单的思路热数据放到 TM 内存中，冷数据放到 TM 的磁盘上。热数据放到内存中可以保障提供更高的性能。

思路很简单，具体如何落地呢？其实 Flink 也不用重复造轮子，直接嵌入一个数据库即可，将数据库嵌入到 TM 的进程内。这里使用一个成熟的嵌入式数据库有非常大的好处，总结如下：

• 成熟的数据库一般性能都比较强，可以提供非常高的查询和写入性能
• 成熟的数据库一般稳定性比较高，不会经常时不时的出现 bug 或出现其他问题
• 成熟的数据库必然考虑了冷热数据的问题，如果使用数据库 Flink 不用考虑这些事情
• 从 0 到 1 打造出一个完全适合 Flink 的高性能状态存储需要非常大的人力成本，所以 Flink 初期先使用业界成熟的数据库是个不错的选择

于是 RocksDBStateBackend 就诞生了。RocksDB 是一个嵌入式、持久化的高性能 KV 存储。Flink 的 RocksDBStateBackend 使用 RocksDB 数据库存储 Flink 的状态数据。RocksDB 会使用内存和 TM 的本地磁盘做为存储介质。

1.3.3 状态存储如何保障高可用？

memory 模式（MemoryStateBackend）和 filesystem 模式（FsStateBackend）会将状态数据存储在 TM 的内存中。rocksdb 模式（RocksDBStateBackend）会将状态数据存储在 TM 本地的 RocksDB 数据库中，也就是存储在 TM 的内存和本地磁盘中。

问题来了，上述三种模式无论将状态数据存储在内存还是磁盘，其实都是在 TM 本地。如果 TM 所在的机器出现故障，就会有丢状态的问题。也就是分布式系统中常见的单点故障。

Flink 做为一个成熟的流计算引擎，对外宣称可以实现 Exactly Once。为了实现业务上的 Exactly Once，Flink 肯定不能丢数据，也就是状态数据必须保障高可靠性。

Flink 依赖 Checkpoint 机制实现了状态数据的高可用。Checkpoint 时，各种 StateBackend 都会进行快照。通俗的来讲，就是将 TM 中的状态数据备份一份到其他存储介质。例如：filesystem 模式（FsStateBackend）在 Checkpoint 时，将 TM 内存中的所有状态数据写一份到 hdfs 上从而保障状态数据的高可用。

到这里就很清晰了：

• 任务运行过程中为了保障高性能，所以状态数据存储在 TM 的内存或者磁盘上
• TM 存在单点故障，所以 Checkpoint 定期将 TM 中的状态写入到外部存储介质（例如 hdfs）来保障高可用

1.3.4 Checkpoint 时数据分别存到哪里了？

memory 模式（MemoryStateBackend）在 Checkpoint 时将状态数据存储在 JM 的内存中，所以仍然存在状态丢失的风险。Flink 官方强调 memory 模式（MemoryStateBackend） 仅适用于调试模式。读到这里相信大家不会在生产环境使用 memory 模式（MemoryStateBackend）。

filesystem 模式（FsStateBackend）在 Checkpoint 时将状态数据存储在文件系统（file system）中，这也是为什么命名为 filesystem 模式（FsStateBackend）。当然这里可以写各种文件系统（file system），例如：HDFS、S3、GFS、NFS、Ceph 等。

注：memory 模式（MemoryStateBackend）和 filesystem 模式（FsStateBackend）在运行过程中的存储结构完全一致，复用的同一套代码（HeapKeyedStateBackend），只是 Checkpoint 时数据写入的位置不同。

rocksdb 模式（RocksDBStateBackend）在 Checkpoint 时与 filesystem 模式类似，可以将状态数据存储在各种文件系统（file system）中。同时 RocksDB 目前支持全量 Checkpoint 和增量 Checkpoint。后续再分析两者区别。

1.4 keyedState 数据存储总结

keyedState 支持三种不同的 StateBackend，三种 StateBackend 都要从两个维度去考虑：

1. 任务运行过程中，StateBackend 将数据存储在哪里？
2. 为了保障高可用，Checkpoint 时StateBackend 将数据存储在哪里？

通过一个表格总结一下：

2、 Operator State 数据存储

Flink 支持三种 StateBackend，但三种 StateBackend 上的 Operator State 在运行期间没有任何区别，都存储在内存中（对应代码 DefaultOperatorStateBackend）。

通常情况下 Operator State 的状态数据量都比较小，所以 Flink 统一都放在了内存中。常见的使用 Operator State 的场景：source offset、buffer 等。

Operator State 比较简单，不多介绍了，放一张表格总结一下：

从上述表格可以看出，filesystem 和 Rocksdb 模式的 Operator State 没有任何区别。如果要在两者之前做选择，请根据 KeyedState 的大小去做选择。

3、 总结

KeyedState 数据存储总结：

Operator State 数据存储总结：