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 恢复的过程: