Flink 源码阅读笔记（10）- State 管理

 * Meta information of the state, including state name, assignment mode, and typeSerializer
 */
private RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo;
/**
 * The internal list the holds the elements of the state
 */
private final ArrayList<S> internalList;
/**
 * A typeSerializer that allows to perform deep copies of internalList
 */
private final ArrayListSerializer<S> internalListCopySerializer;

}

 |
| --- |
从它的成员变量可以看出来，对于 Operator ListState, 其内部就是一个 `ArrayList`。<br />同样，再来看看 `HeapBroadcastState`的实现:
|

public class HeapBroadcastState implements BackendWritableBroadcastState { /**

 * Meta information of the state, including state name, assignment mode, and serializer.
 */
private RegisteredBroadcastStateBackendMetaInfo<K, V> stateMetaInfo;
/**
 * The internal map the holds the elements of the state.
 */
private final Map<K, V> backingMap;
/**
 * A serializer that allows to perform deep copies of internal map state.
 */
private final MapSerializer<K, V> internalMapCopySerializer;

}

 |
| --- |
所以从上面的分析可以看出来，对于所有的 Operator State，都是存储在 TaskManager 的堆内存中的，底层的实现分别对应了 `ArrayList` 和 `HashMap`。
<a name="keyed-state"></a>
### [](https://blog.jrwang.me/2019/flink-source-code-state/#keyed-state)Keyed State
Keyed State 根据底层存储的不同，对应了不同的数据结构和物理存储。 和 `State` 接口相对应，有一个 `InternalKvState` 接口对应状态的内部实现：`State` 接口及其子类对应公共 API，供用户代码调用；`InternalKvState` 及其子类对应内部的具体实现，由内部代码调用。<br />前面提到，内部状态创建的入口在 `KeyedStateBackend#createInternalState` 方法，这个方法在 `AbstractKeyedStateBackend` 中没有提供实现，而是交由不同的状态存储后端自行实现。我们就由该方法作为入口，一窥内部 State 的具体实现。
<a name="heap-keyed-state"></a>
#### [](https://blog.jrwang.me/2019/flink-source-code-state/#heap-keyed-state)Heap Keyed State
`HeapKeyedStateBackend` 将 State 存放在 TaskManager 的堆内存中。
|

HeapKeyedStateBackend extends AbstractKeyedStateBackend { private static final Map, StateFactory> STATE_FACTORIES = Stream.of( Tuple2.of(ValueStateDescriptor.class, (StateFactory) HeapValueState::create), Tuple2.of(ListStateDescriptor.class, (StateFactory) HeapListState::create), Tuple2.of(MapStateDescriptor.class, (StateFactory) HeapMapState::create), Tuple2.of(AggregatingStateDescriptor.class, (StateFactory) HeapAggregatingState::create), Tuple2.of(ReducingStateDescriptor.class, (StateFactory) HeapReducingState::create), Tuple2.of(FoldingStateDescriptor.class, (StateFactory) HeapFoldingState::create) ).collect(Collectors.toMap(t -> t.f0, t -> t.f1));

private final Map<String, StateTable<K, ?, ?>> registeredKVStates;
    public <N, SV, SEV, S extends State, IS extends S> IS createInternalState(
    @Nonnull TypeSerializer<N> namespaceSerializer,
    @Nonnull StateDescriptor<S, SV> stateDesc,
    @Nonnull StateSnapshotTransformFactory<SEV> snapshotTransformFactory) throws Exception {
    StateFactory stateFactory = STATE_FACTORIES.get(stateDesc.getClass());
    if (stateFactory == null) {
        String message = String.format("State %s is not supported by %s",
            stateDesc.getClass(), this.getClass());
        throw new FlinkRuntimeException(message);
    }
    StateTable<K, N, SV> stateTable = tryRegisterStateTable(
        namespaceSerializer, stateDesc, getStateSnapshotTransformFactory(stateDesc, snapshotTransformFactory));
    return stateFactory.createState(stateDesc, stateTable, getKeySerializer());
}

}

 |
| --- |
从上面的代码可以看出来，要创建一个 `InternalKVState`, 首先需要获得一个 `StateTable`， 然后通过 `StateFactory` 接口创建 `InternalKVState`。具体的状态的创建分别对应 `HeapValueState::create`, `HeapListState::create`, `HeapMapState::create`, `HeapAggregatingState::create`, `HeapReducingState::create`。<br />`StateTable` 和 `InternalKVState` 是什么关系呢？在 `HeapXXXState` 的公共父类 `AbstractHeapState` 中可以看出来，对于每一个基于堆内存的 State 的，其底层实际上就是一个 `StateTable`。
|

public abstract class StateTable implements StateSnapshotRestore { protected final InternalKeyContext keyContext; protected RegisteredKeyValueStateBackendMetaInfo metaInfo; //…… }

 |
| --- |
`StateTable` 有三个类型参数，分别对应 key 的类型， namespace 的类型，以及 value 的类型。可以简单地把 `StateTable` 理解成 （key， namespace） -> Value 这样的存储形式。 `StateTable` 有两个具体的实现类，非别为 `NestedMapsStateTable` 和 `CopyOnWriteStateTable`。这两个类分别对应同步和异步模式checkpiont 的情况。其中 `NestedMapsStateTable` 对应同步 checkpoint 的情况，不支持异步快照；而 `CopyOnWriteStateTable` 对应异步 checkpoint 的情况。顾名思义，它的底层提供了 copy-on-write 结构，因而可以支持异步并发的操作。以较为简单的 `NestedMapsStateTable` 为例，可以看出来它是采用两层嵌套的 map 的结构，也就是我们所说的提供了 （key， namespace） -> Value 的映射关系。只有`CopyOnWriteStateTable`，由于提供了 copy-on-write 的支持，实现相对复杂一些，感兴趣的话可以仔细看下它的具体细节，这里就再不分析了。
|

public class NestedMapsStateTable extends StateTable { private final Map>[] state; private final int keyGroupOffset; }

 |
| --- |
接着我们看下 `HeapKeyedStateBackend#tryRegisterStateTable` 的逻辑：
|

class HeapKeyedStateBackend { private StateTable tryRegisterStateTable( TypeSerializer namespaceSerializer, StateDescriptor<?, V> stateDesc, @Nonnull StateSnapshotTransformFactory snapshotTransformFactory) throws StateMigrationException { @SuppressWarnings(“unchecked”) StateTable stateTable = (StateTable) registeredKVStates.get(stateDesc.getName()); TypeSerializer newStateSerializer = stateDesc.getSerializer(); if (stateTable != null) { RegisteredKeyValueStateBackendMetaInfo restoredKvMetaInfo = stateTable.getMetaInfo(); restoredKvMetaInfo.updateSnapshotTransformFactory(snapshotTransformFactory); TypeSerializerSchemaCompatibility namespaceCompatibility = restoredKvMetaInfo.updateNamespaceSerializer(namespaceSerializer); if (namespaceCompatibility.isCompatibleAfterMigration() || namespaceCompatibility.isIncompatible()) { throw new StateMigrationException(“For heap backends, the new namespace serializer must be compatible.”); } restoredKvMetaInfo.checkStateMetaInfo(stateDesc); TypeSerializerSchemaCompatibility stateCompatibility = restoredKvMetaInfo.updateStateSerializer(newStateSerializer); if (stateCompatibility.isIncompatible()) { throw new StateMigrationException(“For heap backends, the new state serializer must not be incompatible.”); } stateTable.setMetaInfo(restoredKvMetaInfo); } else { RegisteredKeyValueStateBackendMetaInfo newMetaInfo = new RegisteredKeyValueStateBackendMetaInfo<>( stateDesc.getType(), stateDesc.getName(), namespaceSerializer, newStateSerializer, snapshotTransformFactory); stateTable = snapshotStrategy.newStateTable(keyContext, newMetaInfo, keySerializer); registeredKVStates.put(stateDesc.getName(), stateTable); } return stateTable; } }

 |
| --- |
这一部分的代码逻辑主要还是基于 `Map<String, StateTable<K, ?, ?>> registeredKVStates` 的 get 和 put 操作来实现，根据 checkpoint 同步或异步配置，创建的 `StateTable` 分别为 `NestedMapsStateTable` 和 `CopyOnWriteStateTable`。<br />接着来看看 `HeapXXXState` 的具体实现。对于 `HeapValueState<K, N, V>`，很明显，其对应在 `StataTable` 中存储的就是状态的具体的值；对于 `HeapListState<K, N, V>`，对应的则是则是 List；对于 `HeapListState<K, N, V>`，对应的是 `Map<V>`。<br />以上几种状态都只是简单的存储和获取操作，但是 `HeapReducingState<K, N, V>` 和 `HeapAggregatingState<K, N, IN, ACC, OUT>` 需要在以前的状态的基础上进行 reduce 和聚合操作。
|

class HeapReducingState extends AbstractHeapMergingState implements InternalReducingState { private final ReduceTransformation reduceTransformation; @Override public void add(V value) throws IOException { if (value == null) { clear(); return; } try { stateTable.transform(currentNamespace, value, reduceTransformation); } catch (Exception e) { throw new IOException(“Exception while applying ReduceFunction in reducing state”, e); } } @Override protected V mergeState(V a, V b) throws Exception { return reduceTransformation.apply(a, b); } static final class ReduceTransformation implements StateTransformationFunction { private final ReduceFunction reduceFunction; ReduceTransformation(ReduceFunction reduceFunction) { this.reduceFunction = Preconditions.checkNotNull(reduceFunction); } @Override public V apply(V previousState, V value) throws Exception { return previousState != null ? reduceFunction.reduce(previousState, value) : value; } } } ``` | | —- |

在上面 HeapReducingState 中，在加入新的值的时候，会调用 ReduceTransformation#apply 方法进行 reduce 操作。在 HeapAggregatingState 中也有类似的逻辑。
在 ReducingStateDescriptor 和 AggregatingStateDescriptor 中提供了对应的 ReduceFunction 和 AggregateFunction。

RocksDB keyed State

//TODO 对于 ROCKSDB 了解不多，后续再加以补充

小结

本文对 Flink 状态管理相关的代码逻辑进行简单的梳理。Flink 中涉及状态管理相关的代码量比较多，而且和 Checkpoint 相关的逻辑杂糅在一起，读起来比较费力。这篇文章主要关注的还是状态的管理，下一篇文章将侧重于 checkpoint 相关的逻辑。

参考

• Flink 的状态管理和检查点机制
• A Deep Dive into Rescalable State in Apache Flink

-EOF-