StreamOperatorStateHandler

在StreamTask启动初始化时通过StreamTaskStateInitializerImpl::streamOperatorStateContext会为每个StreamOperator 创建keyedStatedBackend和operatorStateBackend，在AbstractStreamOperator中有个StreamOperatorStateHandler成员变量，调用AbstractStreamOperator::initializeState方法中会初始化StreamOperatorStateHandler类型的成员变量, StreamOperatorStateHandler对象变量封装了keyedStatedBackend和operatorStateBackend，用于统一管理SteamOperator的状态。

 OperatorChain::initializeStateAndOpenOperators //调用每个Operator的initializeState和Open方法AbstractStreamOperator::initializeState(StreamTaskStateInitializer) StreamTaskStateInitializerImpl::streamOperatorStateContext //此时会创建keyedStatedBackend和operatorStateBackendStreamOperatorStateHandler::new //初始化StreamOperator的stateHandler成员变量，用于状态管理StreamOperatorStateHandler::initializeOperatorStateStateInitializationContextImpl::new //封装DefaultKeyedStateStore和OperatorStateStoreCheckpointedStreamOperator::initializeState(StateInitializationContext)//调用用户定义函数中的initializeState方法，可获取Operator StateStreamingRuntimeContext::setKeyedStateStore

Flink中主要有两种StateBackend：

• HashMapStateBackend //内存
• EmbeddedRocksDBStateBackend //内存+磁盘

每个StreamTask一个StateBackend成员变量，在构造函数中进行初始化，通过用户代码中设置或StateBackendLoader::loadStateBackendFromConfig从配置中加载，默认为HashMapStateBackend。简单起见，以HashMapStateBackend为例剖析创建KeyedStatedBackend和OperatorStateBackend以及处理数据流时是如何使用KeyedState和OperatorState的。

OperatorState

OperatorState创建流程：

OperatorChain::initializeStateAndOpenOperators //调用每个Operator的initializeState和Open方法AbstractStreamOperator::initializeStateStreamTaskStateInitializerImpl::streamOperatorStateContextStreamTaskStateInitializerImpl::operatorStateBackendHashMapStateBackend::createOperatorStateBackend //创建DefaultOperatorStateBackendStreamOperatorStateHandler::new //创建StreamOperatorStateHandlerStreamOperatorStateHandler::initializeOperatorState //调用CheckpointedFunction::initializeStateStateInitializationContextImpl::new //该实例可getOperatorStateStore

使用Operator State的用户业务代码需要实现CheckpointedFunction接口，该接口中有以两个下方法：

void initializeState(FunctionInitializationContext context) throws Exception;void snapshotState(FunctionSnapshotContext context) throws Exception;

其中initializeState方法则会被StreamOperatorStateHandler.initializeOperatorState 调用，在initializeState方法中可使用

FunctionInitializationContext.getOperatorStateStore().getListState(ListStateDescriptor)
DefaultOperatorStateBackend::getListState::newPartitionableListState::new  //内部是ArrayList

因此通过OperatorStateStore获取的ListState内部本质上是一个ArrayList, 业务代码中可以调用add方法向这个内部List添加元素，由StateBackend管理每个Operator State，这样就实现了一个分布式状态管理，借助Checkpoint可以实现状态持久化及容灾恢复。

OperatorStateStore有三个获取状态方法：

<S> ListState<S> getListState(ListStateDescriptor<S> stateDescriptor) throws Exception;
<S> ListState<S> getUnionListState(ListStateDescriptor<S> stateDescriptor) throws Exception;
<K, V> BroadcastState<K, V> getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)throws Exception

KeyedState

KeyedState创建流程如下：

OperatorChain::initializeStateAndOpenOperators //调用每个Operator的initializeState和Open方法AbstractStreamOperator::initializeStateStreamTaskStateInitializerImpl::streamOperatorStateContextStreamTaskStateInitializerImpl::keyedStatedBackendHashMapStateBackend::createKeyedStateBackend //创建HeapKeyedStateBackendHeapKeyedStateBackendBuilder::buildInternalKeyContextImpl::new //用于保存当前正在处理的keyStreamOperatorStateHandler::new //创建StreamOperatorStateHandlerDefaultKeyedStateStore::new //创建DefaultKeyedStateStoreStreamingRuntimeContext::setKeyedStateStore //设置keyedStateStore成员变量AbstractStreamUdfOperator::openFunctionUtils::openFunctionRichFunction::open

KeyedStateStore保存在StreamingRuntimeContext中，使用KeyedState时，用户自定义函数实现RichFunction接口，在open方法中调用getRuntimeContext().getState方法获取状态：

getRuntimeContext().getState() //获取ValueState
DefaultKeyedStateStore::getState
DefaultKeyedStateStore::getPartitionedState
HeapKeyedStateBackend::getPartitionedState
AbstractKeyedStateBackend::getOrCreateKeyedStateLatencyTrackingStateFactory::createStateAndWrapWithLatencyTrackingIfEnabledTtlStateFactory::createStateAndWrapWithTtlIfEnabled //包装TTLHeapKeyedStateBackend::createInternalStateHeapKeyedStateBackend::tryRegisterStateTable //这里很关键，对每个State创建一个StateTableCopyOnWriteStateTable::new//异步快照，这里传递了当前KeyedStateBackend的InternalKeyContextStateTable::new //根据当前Task管理的KeyGroups数量创建StateMap数组CopyOnWriteStateTable::createStateMap //一个KeyGroup一个StateMapCopyOnWriteStateMap::new //存储key及其对应的状态HeapValueState::createHeapValueState::new //有个成员变量指向存储当前state的CopyOnWriteStateMapHeapValueState::setCurrentNamespace  //默认为VoidNamespace

KeyedState有以下几种类型

ValueState<T> getState(ValueStateDescriptor<T> stateProperties) 获取HeapValueStateListState<T> getListState(ListStateDescriptor<T> stateProperties)获取HeapListStateMapState<UK, UV> getMapState(MapStateDescriptor<UK, UV> stateProperties)获取HeapMapStategetAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT> stateProperties)获取HeapAggregatingStategetReducingState(ReducingStateDescriptor<T> stateProperties)获取HeapReducingState

RocksDBStateBackend

EmbeddedRocksDBStateBackend 管理OperatorState与HashMapStateBackend 一样，也是通过DefaultOperatorStateBackend进行管理的。

EmbeddedRocksDBStateBackend 管理KeyedState则是使用RocksDBKeyedStateBackend实现，这样可以借助磁盘加内存进行大状态管理：

RocksDBValueState
RocksDBListState
RocksDBMapState
RocksDBAggregatingState
RocksDBReducingState

总结

Flink内置状态管理是相比其他分布式流式处理系统最大的优势之一，不用借助外部存储组件，就可实现高效可靠的分布式状态管理，极大降低了学习和使用成本。