Flink StateBackend 初探

一个StateBackEnd 包括以下几个部分：
1.CheckPointStreamFactory 构造流用于写出Checkpoint 数据
不同的StateBackEnd会有不同的实现,返回不同的CheckpointStateOutputStream实现，比如 FsStateBackEnd 就会构造文件流, 而MemoryStateBackEnd就会构造ByteArraOutputStream

CheckpointStateOutputStream 会作为IO代理包含在KeyedStateCheckpointOutputStream 和 OperatorStateCheckpointOutputStream内.

KeyedStateCheckpointOutputStream 和 OperatorStateCheckpointOutputStream 分别需要记录额外的状态. KeyedStateCheckpointOutputStream 需要记录每个keyGroup起始在流中的位置, OperatorStateCheckpointOutputStream 需要记录每个partition起始在流中的位置, 这些信息都会体现在对应的StreamStateHandle中.

CheckpointStateOutputStream 定义了 closeAndGetHandle 方法返回了一个 StreamStateHandle 的实现，这个句柄会被序列化传递给JobManager, JobManager 会将句柄作为快照的一部分集中保存，那么在恢复数据的时候就能够通过句柄反向获得InputStream读取数据
具体参考 AbstractStreamOperator.snapshotState
InternalTimerServiceSerializationProxy.write -> HeapInternalTimerService.snapshotTimersForKeyGroup
KeyedStateBackEnd.snapshot OperatorStateBackEnd.snapshot

2.KeyedStateBackEnd
KeyedStateBackEnd 在创建StreamTask 的时候创建，所以一个Task 对应一个KeyedStateBackEnd.

KeyedStateBackEnd 定义了如何注册和生成各种State 包括： ListState, MapState, ValueState, AggregatingState, FoldingState, ReducingState

KeyedStateBackEnd 目前有两种实现： HeapKeyedStateBackend 和 RocksDBKeyedStateBackend. 其中HeapKeyedStateBackend 把状态存储在内部的一个StateTable中，每个State name 对应StateTable 中的一个Entry StateTable 包含三元信息：Key, Namespace, Value. Key和Value 很容易理解, Namespace 目前好像仅仅用于Window 算子，记录了当前的Window 信息， 如果没有Window 会给一个默认的namespace (VoidNamespace.INSTANCE). RocksDBKeyedStateBackend 会根据StateDescription 生成一个RocksDB column family, 然后在每种State get/set 的时候直接对Rocks DB 进行读写操作 *

异步State Snapshot: HeapKeyedStateBackend 和 RocksDBKeyedStateBackend 都支持异步Snapshot, 所谓异步Snapshot 就是起一根独立线程向 CheckpointStateOutputStream 写State 数据. 但是对数据结构有要求，因为在做snapshot 的过程中 state table 本身可能会继续变化. 所以需要在snapshot 开始的时候对数据做一个快照. HeapKeyedStateBackend内部用了CopyOnWriteStateTable保证线程安全性，使数据快照的数据不会corrupt. RocksDBKeyedStateBackend 思路是类似的. snapshot 开始的时候调用RocksDB.snapshot, 然后再通过线程异步向 CheckpointStateOutputStream 写State 数据.

增量 State Snapshot: RocksDBKeyedStateBackend 特有的特性. 具体的实现参考RocksDBIncrementalSnapshotOperation. 这里简单比较一下RocksDBFullSnapshotOperation和RocksDBIncrementalSnapshotOperation. RocksDBFullSnapshotOperation 会完整地读取Snapshot中所有的KV数据，然后向流中写出所有的kvMetadata和kvData. 返回的StateHandle是KeyGroupsStateHandle, 和HeapKedStateBackend一致. 而RocksDBIncrementalSnapshotOperation则会遍历RocksDB checkpoint目录下的所有文件. 每次做Checkpoint的时候，RocksDBKeyedStateBackend会记录当前checkPointId对应的RocksDB ssd文件.这样在做一次新的Checkpoint的时候就可以比对文件获取是否有新的数据文件.原有的数据文件不用再写而是直接返回一个PlaceholderStreamStateHandle. Checkpoint不是逐条遍历KV写出,而是直接向流中写出RocksDB数据文件的数据. 返回的StateHandle是IncrementalKeyedStateHandle其中包含了一组RocksDB数据文件的句柄.
数据恢复的过程也同样需要区分full/incremental. 分别对应RocksDBFullRestoreOperation和RocksDBIncrementalRestoreOperation

3.OperatorStateBackEnd
主要管理OperatorState. 目前只有一种实现: DefaultOperatorStateBackend. 构造出一个 PartitionableListState (属于ListState). 这是一个In Memory的实现. Add 操作追加到
内存的一个List中. Snapshot 的过程和KeyedStateBackEnd大同小异，这里就不再赘述.

StateBackend 的类结构:

State 恢复的过程: