文章目录1.基于MemoryStateBackend创建KeyedStateBackend1.1. 状态初始化1.2. 创建状态2. 基于MemoryStateBackend创建OperatorStateBackend3.基于MemoryStateBackend创建CheckpointStorage在Flink中默认的StateBackend实现为MemoryStateBackend本文以MemoryStateBackend为例说明StateBackend的设计与实现。本文介绍MemoryStateBackend中如下三个主要组件的创建过程HeapKeyedStateBackendOperatorStateBackendMemoryBackendCheckpointStorageFsStateBackend和RocksDBStateBackend这两种状态后端存储的实现功能和MemoryStateBackend类似区别在于内部创建的KeyedStateBackend和CheckpointStorage。1.基于MemoryStateBackend创建KeyedStateBackend1.1. 状态初始化AbstractStreamOperator.keyedStatedBackend()方法定义了创建和初始化KeyedStatedBackend的逻辑具体如下。protectedKAbstractKeyedStateBackendKkeyedStateBackend(TypeSerializerKkeySerializer,StringoperatorIdentifierText,PrioritizedOperatorSubtaskStateprioritizedOperatorSubtaskStates,CloseableRegistrybackendCloseableRegistry,MetricGroupmetricGroup)throwsException{if(keySerializernull){returnnull;}StringlogDescriptionkeyed state backend for operatorIdentifierText;//1.TaskInfotaskInfoenvironment.getTaskInfo();finalKeyGroupRangekeyGroupRangeKeyGroupRangeAssignment.computeKeyGroupRangeForOperatorIndex(taskInfo.getMaxNumberOfParallelSubtasks(),taskInfo.getNumberOfParallelSubtasks(),taskInfo.getIndexOfThisSubtask());// 确保恢复状态过程中构建的数据流被关闭CloseableRegistrycancelStreamRegistryForRestorenewCloseableRegistry();backendCloseableRegistry.registerCloseable(cancelStreamRegistryForRestore);// 创建BackendRestorerProcedureBackendRestorerProcedureAbstractKeyedStateBackendK,KeyedStateHandlebackendRestorernewBackendRestorerProcedure((stateHandles)-stateBackend.createKeyedStateBackend(environment,environment.getJobID(),operatorIdentifierText,keySerializer,taskInfo.getMaxNumberOfParallelSubtasks(),keyGroupRange,environment.getTaskKvStateRegistry(),TtlTimeProvider.DEFAULT,metricGroup,stateHandles,cancelStreamRegistryForRestore),backendCloseableRegistry,logDescription);try{returnbackendRestorer.createAndRestore(prioritizedOperatorSubtaskStates.getPrioritizedManagedKeyedState());}finally{if(backendCloseableRegistry.unregisterCloseable(cancelStreamRegistryForRestore)){IOUtils.closeQuietly(cancelStreamRegistryForRestore);}}}获取当前Task的TaskInfo并基于TaskInfo的参数创建KeyGroupRange表示当前Task实例中存储的Key分组区间。创建CloseableRegistry并注册到backendCloseableRegistry中用于确保在任务取消的情况下关闭在恢复状态过程中构造的数据流。创建BackendRestorerProcedure提供了stateBackend.createKeyedStateBackend()方法也包含恢复历史状态数据的方法。创建KeyedStateBackend同时对状态数据进行恢复。prioritizedOperatorSubtaskStates是从TaskStateManager中根据OperatorID获取的算子历史状态通过prioritizedOperatorSubtaskStates获取当前算子的PrioritizedManagedKeyedState并基于这些状态数据恢复算子的状态。1.2. 创建状态接下来我们看MemoryStateBackend.createKeyedStateBackend()方法的具体实现。publicKAbstractKeyedStateBackendKcreateKeyedStateBackend(Environmentenv,JobIDjobID,StringoperatorIdentifier,TypeSerializerKkeySerializer,intnumberOfKeyGroups,KeyGroupRangekeyGroupRange,TaskKvStateRegistrykvStateRegistry,TtlTimeProviderttlTimeProvider,MetricGroupmetricGroup,NonnullCollectionKeyedStateHandlestateHandles,CloseableRegistrycancelStreamRegistry)throwsBackendBuildingException{// 获取TaskStateManager实例TaskStateManagertaskStateManagerenv.getTaskStateManager();// 创建HeapPriorityQueueSetFactory实例HeapPriorityQueueSetFactorypriorityQueueSetFactorynewHeapPriorityQueueSetFactory(keyGroupRange,numberOfKeyGroups,128);// 创建HeapKeyedStateBackendBuilder实例HeapKeyedStateBackendreturnnewHeapKeyedStateBackendBuilder(kvStateRegistry,keySerializer,env.getUserClassLoader(),numberOfKeyGroups,keyGroupRange,env.getExecutionConfig(),ttlTimeProvider,stateHandles,AbstractStateBackend.getCompressionDecorator(env.getExecutionConfig()),taskStateManager.createLocalRecoveryConfig(),priorityQueueSetFactory,isUsingAsynchronousSnapshots(),cancelStreamRegistry).build();}从environment参数中获取TaskStateManager实例创建HeapPriorityQueueSetFactory实例用于生成HeapPriorityQueueSet优先级队列存储TimerHeapInternalTimer等数据。调用HeapKeyedStateBackendBuilder.build()方法创建HeapKeyedStateBackend。2. 基于MemoryStateBackend创建OperatorStateBackend和创建KeyedStateBackend的过程相似AbstractStreamOperator.operatorStateBackend()方法实现了创建OperatorStateBackend的方法。protectedOperatorStateBackendoperatorStateBackend(StringoperatorIdentifierText,PrioritizedOperatorSubtaskStateprioritizedOperatorSubtaskStates,CloseableRegistrybackendCloseableRegistry)throwsException{StringlogDescriptionoperator state backend for operatorIdentifierText;CloseableRegistrycancelStreamRegistryForRestorenewCloseableRegistry();backendCloseableRegistry.registerCloseable(cancelStreamRegistryForRestore);BackendRestorerProcedureOperatorStateBackend,OperatorStateHandlebackendRestorernewBackendRestorerProcedure((stateHandles)-stateBackend.createOperatorStateBackend(environment,operatorIdentifierText,stateHandles,cancelStreamRegistryForRestore),backendCloseableRegistry,logDescription);try{returnbackendRestorer.createAndRestore(prioritizedOperatorSubtaskStates.getPrioritizedManagedOperatorState());}finally{if(backendCloseableRegistry.unregisterCloseable(cancelStreamRegistryForRestore)){IOUtils.closeQuietly(cancelStreamRegistryForRestore);}}}创建CloseableRegisty确保在任务取消的情况下能够关闭在恢复状态时构造的数据流。创建BackendRestorerProcedure封装了stateBackend.createOperatorStateBackend()方法并包含恢复历史状态数据的操作。创建OperatorStateBackend并恢复状态数据。其中prioritizedOperatorSubtaskStates是从TaskStateManager中根据OperatorID获取的算子专有历史状态可以通过prioritizedOperatorSubtaskStates获取当前算子中的PrioritizedManagedOperatorState并基于这些状态数据恢复OperatorStateBackend中算子的状态。3.基于MemoryStateBackend创建CheckpointStorage在createCheckpointStorage()方法中直接创建MemoryBackendCheckpointStorage实例并返回没有涉及太多的流程publicCheckpointStoragecreateCheckpointStorage(JobIDjobId)throwsIOException{returnnewMemoryBackendCheckpointStorage(jobId,getCheckpointPath(),getSavepointPath(),maxStateSize);}参考《Flink设计与实现核心原理与源码解析》–张利兵
