[Flink State] State究竟保存在哪里？
1.前言
从源码解析State的保存过程，上一篇从task和operator出发说明了保存state的过程，到最后是由算子调用snapshot方法，进行state的快照操作。那么state究竟保存在哪里？
<1> State Backend简介
- 状态后端定义了流式应用程序状态如何存储和checkpoint的。不同的状态后端以不同的方式来存储其状态，并且使用不同的数据结构来保存正在运行的应用程序的状态。
- 例如，MemoryStateBackend将工作state保存在TaskManager的内存中，并将checkpoint数据存储在JobManager的内存中。后端是轻量级的，没有其他依赖关系，但是可用性不高并且仅支持小状态。
- FsStateBackend将工作state保存在TaskManager的内存中，并将checkpoint数据存储在文件系统中。
- RocksDBStateBackend将工作state保存在RocksDB中，并且默认将checkpoint数据存在文件系统中，类似FsStateBackend。
关于 Raw Bytes Storage and Backends
- StateBackend提供服务给raw bytes storage，keyed state和operator state。
- raw bytes存储（通过CheckpointStreamFactory）是一个基础服务以可容错的方式简单存储。该服务通过JobManager来存储checkpoint数据和恢复元数据，通常也可以提供给keyed-和operator状态后端来存储checkpoint数据。
- StateBackend创建的AbstractKeyedStateBackend和OperatorStateBackend定义如何保持keyed和operator状态。他们也定义了如何将state给checkpoint（经常使用raw bytes storage通过CheckpointStreamFactory）。但是也有可能例如Keyed state backend只是实现了键/值存储，并且不需要在原始字节存储中存储任何内容检查点。
总结：StateBackend主要是针对raw bytes storage（即checkpoint），keyed state和operator state来提供功能的，其中checkpoint数据的存储则是通过CheckpointStreamFactory，而state存储，针对keyedState是通过AbstractKeyedStateBackend，针对operatorState是通过OperatorStateBackend。
<2>
| - | MemoryStateBackend | FsStateBackend | RocksDBStateBackend |
|---|---|---|---|
| CheckpointStream | MemCheckpointStreamFactory | FsCheckpointStreamFactory | FsCheckpointStreamFactory |
| SavepointStream | MemCheckpointStreamFactory | FsSavepointStreamFactory | FsSavepointStreamFactory |
| KeyedStateBackend | HeapKeyedStateBackend | HeapKeyedStateBackend | RocksDBKeyedStateBackend |
| OperatorStateBackend | DefaultOperatorStateBackend | DefaultOperatorStateBackend | DefaultOperatorStateBackend |
RocksDBStateBackend的构造函数可以传入一个AbstractStateBackend，否则默认采用FsStateBackend。
可以看到，从OperatorState的角度来讲，目前Flink只有一个实现，即DefaultOperatorStateBackend，它将List风格的State保存在内存中。
从KeyedState的角度来讲，目前有两种实现,HeapKeyedStateBackend将state保存在内存中，而RocksDbKeyedStateBackend将State保存在TM本地的RocksDB中。相对而言，前者在内存中，速度会快，效率高，但一方面会限制state的大小，另一方面也会造成JVM自己的内存问题；后者在本地文件中，就会涉及序列化和反序列化，效率不及前者，但可以保存的state的大小会很大。
从checkpoint和savepoint的角度来看，Memory工厂方法都保存在内存中，显然不能在生产环境使用，而Fs工厂方法和RocksDb工厂方法，则统一都放在文件系统中，比如HDFS。
2.源码分析
从上图中3,4两行可以看到，具体用来存储state的有三种HeapKeyedStateBackend，RocksDBKeyedStateBackend和DefaultOperatorStateBackend。
DefaultOperatorStateBackend
（1）
通过上面可以看到，状态是以Map方式来存储的
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不一样。
