flink运行计算任务的过程中，会有很多中间处理过程。在整个任务运行的过程中，中间存在着多个临时状态，比如说某些数据正在执行一个operator，但是只处理了一半数据，另外一般还没来得及处理，这也是一个状态。
假设运行过程中，由于某些原因，任务挂掉了，或者flink中的多个task中的一个task挂掉了，那么它在内存中的状态都会丢失，如果这时候我们没有存储中间计算的状态，那么就意味着重启这个计算任务时，需要从头开始将原来处理过的数据重新计算一遍。如果存储了中间状态，就可以恢复到中间状态，并从该状态开始继续执行任务。这就是状态管理的意义。所以需要一种机制去保存记录执行过程中的中间状态，这种机制就是状态管理机制。
flink中包括两种基础状态：keyed state（keyed状态）和operator state（operator状态）
keyed状态总是与key一起，且只能用在keyedStream中。这个状态是跟特定的key绑定的，对KeyedStream流上的每一个key，可能都对应一个state。唯一组合成（operator--key，state）的形式。
与Keyed State不同，Operator State跟一个特定operator的一个并发实例绑定，整个operator只对应一个state。相比较而言，在一个operator上，可能会有很多个key，从而对应多个keyed state。而且operator state可以应用于非keyed stream中。
举例来说，Flink中的Kafka Connector，就使用了operator state。它会在每个connector实例中，保存该实例中消费topic的所有(partition, offset)映射。
所有state的存在形式有两种：managed（委托管理）和raw（原始）。
托管方式就是状态管理由flink提供的框架进行管理，通过flink状态管理框架提供的接口，来更新和管理状态的值。这里面包括用于存储状态数据的数据结构，现成的包装类等。
原始方式就是由用户自行管理状态具体的数据结构，框架在做checkpoint的时候（checkpoint是flink进行状态数据持久化存储的机制），使用byte[]来读写状态内容，对其内部数据结构一无所知。
通常在DataStream上的状态推荐使用托管的状态，当实现一个用户自定义的operator时，会使用到原始状态。一般来说，托管状态用的比较多。
flink提供了很多存储state数据的类，类继承图如下：
图 1.4 flink--状态管理类继承图
可用于存储状态数据的类如下：
ValueState<T>*这保留了一个可以更新和检索的值。即类型为T的单值状态。这个状态与对应的key绑定，是最简单的状态了。它可以通过update方法更新状态值，通过value()方法获取状态值。
ListState<T>*这保存了一个元素列表，即key上的状态值为一个列表。可以追加元素并检索Iterable所有当前存储的元素。可以通过add方法往列表中附加值；也可以通过get()方法返回一个Iterable<T>来遍历状态值。
ReducingState<T>*这保留了一个值，该值表示添加到状态的所有值的聚合。接口类似于ListState，但是添加的元素使用add(T)。每次调用add方法添加值的时候，会调用reduceFunction，最后合并到一个单一的状态值。
AggregatingState<IN, OUT>*这保留了一个值，该值表示添加到状态的所有值的聚合。和ReducingState不同的是，聚合类型可能与添加到状态的元素类型不同
FoldingState<T, ACC>*这保留了一个值，该值表示添加到状态的所有值的聚合。违背ReducingState，聚合类型可能与添加到状态的元素类型不同。接口类似于ListState但是添加的元素使用add(T)使用指定的FoldFunction.这个基本弃用，请用AggregatingState代替
MapState<UK, UV>*它保存了一个映射列表。可以将键值对放入状态并检索Iterable所有当前存储的映射。映射使用put(UK, UV)或putAll(Map<UK, UV>)添加元素。使用get(UK)获取元素。映射、键和值的可迭代视图可以分别使用entries(), keys()和values()需要注意的是，以上所述的State对象，仅仅用于与状态进行交互（更新、删除、清空等），而真正的状态值，有可能是存在内存、磁盘、或者其他分布式存储系统中。相当于我们只是持有了这个状态的句柄(state handle)。
接下来看下，我们如何得到这个状态句柄。Flink通过StateDescriptor来定义一个状态。这是一个抽象类，内部定义了状态名称、类型、序列化器等基础信息。与上面的状态类型对应。如下：
ValueStateDescriptor
ListStateDescriptor
ReducingStateDescriptor
FoldingStateDescriptor
AggregatingStateDescriptor
MapStateDescriptor以keyed state为例，
1、首先，普通Function接口是不支持状态管理的，也就是一般故障的情况下，状态并没有保存下来，后面需要将所有数据进行重新计算。如果需要支持状态管理，那么我们需要继承实现 RichFunction类。基本常用的function，flink都再封装了对应的RichFunction接口给我们使用，比如普通function中的MapFunction，对应的RichFunction抽象类为RichMapFunction。命名方式对应关系很简单，基本就是 xxxFunciotn -->RichxxxFunction。
2、接着，需要在覆盖实现RichFunction中的对应的算子方法（如map、flatMap等），里面需要实现算子业务逻辑，并将对keyed state进行更新、管理。然后还要重写open方式，用于获取状态句柄。
下面使用ValueState为例，实现RichFlatMapFunction接口：
public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
/**
* ValueState状态句柄. 第一个值为count，第二个值为sum。
*/
private transient ValueState<Tuple2<Long, Long>> sum;
@Override
public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
// 获取当前状态值
Tuple2<Long, Long> currentSum = sum.value();
// 更新
currentSum.f0 += 1;
currentSum.f1 += input.f1;
// 更新状态值
sum.update(currentSum);
// 如果count >=2 清空状态值，重新计算
if (currentSum.f0 >= 2) {
out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
sum.clear();
}
}
@Override
public void open(Configuration config) {
ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
new ValueStateDescriptor<>(
"average", // 状态名称
TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}), // 当个状态数据的类型，这里是tuple,也就是元祖
Tuple2.of(0L, 0L)); // 状态默认值
//获取状态句柄
sum = getRuntimeContext().getState(descriptor);
}
}
// 主程序
env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L))
.keyBy(0)
.flatMap(new CountWindowAverage())
.print();
// the printed output will be (1,4) and (1,5)以2.2 中的例子为例，看状态句柄是如何获取的。
首先从这里入手：
sum = getRuntimeContext().getState(descriptor);首先RichFlatMapFunction类中没有这个方法，再到父类AbstractRichFunction，如下：
public abstract class AbstractRichFunction implements RichFunction, Serializable {
private static final long serialVersionUID = 1L;
private transient RuntimeContext runtimeContext;
public AbstractRichFunction() {
}
public void setRuntimeContext(RuntimeContext t) {
this.runtimeContext = t;
}
//返回RuntimeContext类的对象
public RuntimeContext getRuntimeContext() {
if (this.runtimeContext != null) {
return this.runtimeContext;
} else {
throw new IllegalStateException("The runtime context has not been initialized.");
}
}
...........................
}getRuntimeContext()这个方法在这里实际上是返回了StreamingRuntimeContext这个子类对象。所以调用了RichFlatMapFunction.getRuntimeContext().getState方法，最终会调用StreamingRuntimeContext.getState方法：
/* StreamingRuntimeContext */
public class StreamingRuntimeContext extends AbstractRuntimeUDFContext {
...........
public <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties) {
//这里获取到KeyedStateStore对象
KeyedStateStore keyedStateStore = this.checkPreconditionsAndGetKeyedStateStore(stateProperties);
stateProperties.initializeSerializerUnlessSet(this.getExecutionConfig());
return keyedStateStore.getState(stateProperties);
}
...............
}接着看this.checkPreconditionsAndGetKeyedStateStore，
private KeyedStateStore checkPreconditionsAndGetKeyedStateStore(StateDescriptor<?, ?> stateDescriptor) {
Preconditions.checkNotNull(stateDescriptor, "The state properties must not be null");
KeyedStateStore keyedStateStore = this.operator.getKeyedStateStore();
Preconditions.checkNotNull(keyedStateStore, "Keyed state can only be used on a 'keyed stream', i.e., after a 'keyBy()' operation.");
return keyedStateStore;
}再接着看 this.operator.getKeyedStateStore();，
/*AbstractStreamOperator*/
.....
public KeyedStateStore getKeyedStateStore() {
return this.keyedStateStore;
}
.....返回了AbstractStreamOperator.keyedStateStore变量。这个变量的初始化在AbstractStreamOperator.initializeState方法中：
public final void initializeState() throws Exception {
TypeSerializer<?> keySerializer = this.config.getStateKeySerializer(this.getUserCodeClassloader());
StreamTask<?, ?> containingTask = (StreamTask)Preconditions.checkNotNull(this.getContainingTask());
CloseableRegistry streamTaskCloseableRegistry = (CloseableRegistry)Preconditions.checkNotNull(containingTask.getCancelables());
//创建 StreamTaskStateInitializerImpl 对象
StreamTaskStateInitializer streamTaskStateManager = (StreamTaskStateInitializer)Preconditions.checkNotNull(containingTask.createStreamTaskStateInitializer());
//这里创建一个当前operator的state context对象
StreamOperatorStateContext context = streamTaskStateManager.streamOperatorStateContext(this.getOperatorID(), this.getClass().getSimpleName(), this, keySerializer, streamTaskCloseableRegistry);
//通过state context这个上下文对象获取keyed state和operator state的 backend配置
this.operatorStateBackend = context.operatorStateBackend();
this.keyedStateBackend = context.keyedStateBackend();
//初始化keyedStateStore，将从配置中创建的statebackend封装到默认的statebackend
if (this.keyedStateBackend != null) {
this.keyedStateStore = new DefaultKeyedStateStore(this.keyedStateBackend, this.getExecutionConfig());
}
this.timeServiceManager = context.internalTimerServiceManager();
CloseableIterable<KeyGroupStatePartitionStreamProvider> keyedStateInputs = context.rawKeyedStateInputs();
CloseableIterable operatorStateInputs = context.rawOperatorStateInputs();
try {
StateInitializationContext initializationContext = new StateInitializationContextImpl(context.isRestored(), this.operatorStateBackend, this.keyedStateStore, keyedStateInputs, operatorStateInputs);
this.initializeState(initializationContext);
} finally {
closeFromRegistry(operatorStateInputs, streamTaskCloseableRegistry);
closeFromRegistry(keyedStateInputs, streamTaskCloseableRegistry);
}
}在containingTask.createStreamTaskStateInitializer()，也就是StreamTask.createStreamTaskStateInitializer()这里有初始化操作，看看
public StreamTaskStateInitializer createStreamTaskStateInitializer() {
return new StreamTaskStateInitializerImpl(this.getEnvironment(), this.stateBackend, this.timerService);
}创建了一个StreamTaskStateInitializerImpl对象，里面传入了 this.stateBackend这个参数，估计就是获取用户指定的stateBackend类型，追踪this.stateBackend这个参数在哪初始化的，找到了StreamTask.invoke这个方法，如下：
public final void invoke() throws Exception {
..........
this.stateBackend = this.createStateBackend();
............
}调用了this.createStateBackend()，即StreamTask.createStateBackend()继续看
private StateBackend createStateBackend() throws Exception {
//从application中获取已经创建的statebackend object
StateBackend fromApplication = this.configuration.getStateBackend(this.getUserCodeClassLoader());
//加载指定的 statebackend类，并返回statebackend对象
return StateBackendLoader.fromApplicationOrConfigOrDefault(fromApplication, this.getEnvironment().getTaskManagerInfo().getConfiguration(), this.getUserCodeClassLoader(), LOG);
}继续看this.configuration.getStateBackend, 即StreamConfing.getStateBackend
public StateBackend getStateBackend(ClassLoader cl) {
try {
//从application的config中获取"statebackend"这个key对应的object，也就是已创建的backend对象。
return (StateBackend)InstantiationUtil.readObjectFromConfig(this.config, "statebackend", cl);
} catch (Exception var3) {
throw new StreamTaskException("Could not instantiate statehandle provider.", var3);
}
}从application的config中获取"statebackend"这个key对应的object，也就是已创建的backend对象。如果application不存在，则从配置中读取应该使用的backend类型，然后加载。看这个方法StateBackendLoader.fromApplicationOrConfigOrDefault。这个方法就是从config配置中读取“state.backend”这个配置项值，然后加载对应的backend。
到最后，要记住this.keyedStateStore = new DefaultKeyedStateStore(this.keyedStateBackend, this.getExecutionConfig());这个是 this.keyedStateStore封装后的对象。
接着回到StreamingRuntimeContext.getState中，看到 this.keyedStateStore.getState，这一行，调用的是DefaultKeyedStateStore.getState,其实就再次进行初始化而已，这里不展开。
首先，我们这里使用的是ValueState，它的直接实现子类是HeapValueState。它的存储数据结构，在它的父类AbstractHeapState中，以StateTable<K, N, SV> stateTable的形式存在的，其中K代表Key的类型，N代表state的namespace（这样属于不同namespace的state可以重名），SV代表state value的类型。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
