[Flink State] State究竟保存在哪里？

Posted 2023-03-09

参考技术A

从源码解析State的保存过程 ，上一篇从task和operator出发说明了保存state的过程，到最后是由算子调用snapshot方法，进行state的快照操作。那么state究竟保存在哪里？

<1> State Backend简介

关于 Raw Bytes Storage and Backends

总结： StateBackend主要是针对raw bytes storage（即checkpoint），keyed state和operator state来提供功能的，其中checkpoint数据的存储则是通过CheckpointStreamFactory，而state存储，针对keyedState是通过AbstractKeyedStateBackend，针对operatorState是通过OperatorStateBackend。

<2>

RocksDBStateBackend的构造函数可以传入一个AbstractStateBackend，否则默认采用FsStateBackend。

可以看到，从OperatorState的角度来讲，目前Flink只有一个实现，即DefaultOperatorStateBackend，它将List风格的State保存在内存中。
从KeyedState的角度来讲，目前有两种实现,HeapKeyedStateBackend将state保存在内存中，而RocksDbKeyedStateBackend将State保存在TM本地的RocksDB中。相对而言，前者在内存中，速度会快，效率高，但一方面会限制state的大小，另一方面也会造成JVM自己的内存问题；后者在本地文件中，就会涉及序列化和反序列化，效率不及前者，但可以保存的state的大小会很大。

从checkpoint和savepoint的角度来看，Memory工厂方法都保存在内存中，显然不能在生产环境使用，而Fs工厂方法和RocksDb工厂方法，则统一都放在文件系统中，比如HDFS。

从上图中3,4两行可以看到，具体用来存储state的有三种HeapKeyedStateBackend，RocksDBKeyedStateBackend和DefaultOperatorStateBackend。

DefaultOperatorStateBackend
（1）

operator的ListSate的实现类PartitionableListState， OperatorState都保存在内存中，本质上还是一个ArrayList。

（2）snapshot方法

该snapshotStrategy是AbstractSnapshotStrategy<OperatorStateHandle>，而AbstractSnapshotStrategy有三种实现类：

DefaultOperatorStateBackendSnapshotStrategy中的snapshot方法：
该snapshot方法中，主要是对 registeredOperatorStates 和 registeredBroadcastStates 的snapshot。

第一步：

针对所有注册的state进行deepCopy，为了防止在checkpoint的时候数据结构又被修改，deepCopy其实是通过序列化和反序列化的过程；

第二步：
异步写入State和MetaInfo，先创建CheckpointStateOutputStream，通过调用factory的createCheckpointStateOutputStream方法，这个factory是哪种类型的呢？这个是由定义的状态后端所决定的。之后会返回相应的OperatorstateHandle用作restore的过程。

第三步：
在StreamTask触发checkpoint的时候会将一个Task中所有的operator触发一次snapshot，触发部分就是上面1，2两个步骤，其中第二步是会返回一个RunnableFuture，在触发之后会提交一个AsyncSnapshotCallable异步任务，会阻塞一直等到checkpoint的Future，其实就是去调用这个方法AbstractAsyncIOCallable, 直到完成之后OperatorState会返回一个OperatorStateHandle,这个地方和后文的keyedState返回的handle不一样。