这篇文档解释了如何在开发应用时，使用Flink的state。
Keyed State and Operator State
在Flink有两种基本的state：keyed state 与 operator state
Keyed state
keyed State 是与key关联的，只能在应用于keyedStream上的function与operator中使用。
你可以认为keyed state是分区后每个key都有自己的一个state的operator state。每一个keyed state逻辑上都与<parallel-operator-instance, key> 一一对应，又由于每一个key仅存在于某个特定的key operator实例，因此也可以说与<operator, key>一一对应。
keyed state 然后会被组织为key group。Key Group是一个原子性的单位，可以被flink用于重分配key state。key group的个数与job最大并行度相同（注：也就是有可能某个key operator实例持有多个key group）。在执行期间，每一个key operator实例都可以对其持有的一个或多个key group中的多个key state进行操作。
Operator state
对于 operator state（或者 non-keyed state）来说，每个operator state 都与一个operator 实例一一对应。Kafka connector是一个很好的例子。kafka consumer的每一个实例，都维持一个map（map类型的state），包含topic partitions以及offset。
当opetaor的并行度发生改变时，operator state接口支持对其state的重分配。有多种方案实现这种重分配。
（注：他俩的区别可以这么理解：Operator State:每个应用于non-key stream的operator实例都可以有自己的state，operator 与 state对应。而Keyed State：每个应用于keyedStream 的operator实例对应的是key group，key group是由多个key的state组成的，而且key state还可以在function中）
Raw and Managed State
keyed state 与 operator state 都存在两种形式： managed 与 raw。
managed state 代表flink可以在运行时控制的数据结构，数据结构如内部的hash table,RocksDB。managed state的举例，如：ValueState,LIstState等。Flink的运行时会对state进行编码，并且写入checkpoint。
raw state 是那些在operator中使用自己的数据结构的state。当checkpoint时，他们仅会写一个字节序列到checkpoint中。Flink并不知道state的数据结构，仅仅是看到这些字节。
所有的function中都可以使用managed state，而raw state仅可以在实现operator时使用。推荐使用managed state，因为当job的并发度发生改变时，flink可以自动的对managed state进行重分配，并且使用managed state有更好的内存管理。
注意：如果你的managed state需要自定义的序列化逻辑，可以参阅corresponding guide来确保兼容性。Flink的默认序列化器不需要特殊处理。
Using Managed Keyed State
managed key state 接口提供了访问不同类型state能力，这些state都对应于输入流中数据的key。这意味着，managed key state只能在 keyedStream 中使用，一般可以通过 stream.keyBy(…)来创建 keyedStream。
现在，我们会先看一下哪些类型的state可用，然后会展示如何在代码中使用。下面是可用的基本state：
- ValueState<T>:维持一个可以被更新与获取的value（就如上文说的，该state对应于一个key，因此对于一个operator实例来说，可以获取到其中各个key的对应的state），state的值可以通过 update(T) 来更新，可以通过 T value()来获取。
- ListState<T>:这个state会维持一个list。你可以追加元素，也可以获取到当前存储的所有元素的 Iterable。通过 add(T) / addAll(List<T>) 来添加元素，通过 Iterable<T> get() 方法获取迭代器。你也可以使用 update(List<T>) 来覆盖state。
- ReducingState<T>:这个state维持一个值，该值代表所有添加到这个状态的值的聚合。接口和ListState类似，但是使用 add(T)添加元素，会导致该值进入 ReduceFunction 执行聚合操作。
- AggregatingState<IN,OUT>:这个state维持一个值，该值代表所有添加到这个状态的值的聚合。与ReducingState不一样的是，聚合后的值的类型可以和输入值的类型不同。接口和ListState类似，但是使用 add(T)添加元素，会导致该值进入 AggregateFunction 执行聚合操作。
- FoldingState<T,ACC):这个state在1.4就弃用了，在后面的版本中会完全删除，可以使用 AggregatingState代替
- MapState<UK,UV>:这个state维持了一个map。你可以将key-value对put到state中也可以获取当前map的Iterable。可以使用 put(UK,UV)/putAll(Map<UK,UV>)来添加map。获取某个user key的值，可以使用 get(UK).Iterable 可以使用 entries(),keys(),values()方法。
所有类型的state都有clear()方法，该方法会清除当前key对应的状态。
第一件重要的事情是，这些state对象都是仅用来连接state的，state不一定存储在内存中，也有可能在磁盘上或者其他地方。第二件重要的事情是，你从state中获取到的value取决于输入流中的key。因此如果你在自定义函数中调用了state的方法获取值时，在不同的key时，返回的结果是不同的。
为了手动获取state，你必须创建 StateDescriptor。这个类会保存这个state的名称，state中value的类型，以及可能会有的自定义函数，如 ReduceFunction（你可以定义好几个state，使用不同的名字命名，这样可以通过名称来获取它们）。取决于你想要使用何种类型的state，你可以选择创建如下的descriptor：ValueStateDescriptor, ListStateDescriptor, ReducingStateDescriptor, FoldingStateDescriptor 或者 MapStateDescriptor。
state 可以通过 RuntimeContext 来获取，因此只有在 rich function中才可以使用state。RichFunction中的RuntimeContext有如下方法可以获取state：
- ValueState<T> getState( ValueStateDescriptor<T>)
- ReducingState<T> getReducingState(ReducingStateDescriptor<T>)
- ListState<T> getListState(ListStateDescriptor<T>)
- AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT>)
- FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC>)
- MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)
下面的示例使用 FlatMapFunctino 展示了如何使用上面说的特性：
public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
/**
* The ValueState handle. The first field is the count, the second field a running sum.
*/
private transient ValueState<Tuple2<Long, Long>> sum;
@Override
public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
// access the state value
Tuple2<Long, Long> currentSum = sum.value();
// update the count
currentSum.f0 += 1;
// add the second field of the input value
currentSum.f1 += input.f1;
// update the state
sum.update(currentSum);
// if the count reaches 2, emit the average and clear the state
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", // the state name
TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}), // type information
Tuple2.of(0L, 0L)); // default value of the state, if nothing was set
sum = getRuntimeContext().getState(descriptor);
}
}
// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)
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)
这个示例中，使用tuple的第一个field作为key（在示例中，只有一个相同的key：1），在function中，存储该key出现的次数以及计算总和，放入Valuestate中。一旦次数大于等于2，就会计算并发射平均值，然后清空state，这样我们就会再次从0开始。注意的是，如果我们的输入有多个key的话，对于不同的key会有不同的state value。
State Time-To-Live(TTL)
任何类型的state都可以指定ttl。当配置了TTL并且state value过期后，存储的value会已一种高效的方式清除，下面会详细介绍。
state的集合类型支持per-entry TTL，这意味着list中的元素或者map中的entry的过期时间是相互独立的。
为了使用state的TTL功能，必须要构造一个 StateTtlConfig 的配置对象。然后就可以通过向state descriptor中传configration的方式实现对应state的TTL功能。
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build();
ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("text state", String.class);
stateDescriptor.enableTimeToLive(ttlConfig);
configuration有几个需要注意的点：
newBuilder 方法的第一个参数是固定的，就是 time-to-live 的值
update type 配置什么时候state的TTL被刷新（默认是 onCreateAndWrite）
- StateTtlConfig.UpdateType.OnCreateAndWrite – 仅当创建或写入时刷新
- StateTtlConfig.UpdateType.OnReadAndWrite – 创建，写入或读取时都可以刷新
state visibility 配置了当过期数据未被清除时，读取该数据是否会返回数据的值。（默认是 NeverRetrunExpired）
- StateTtlConfig.StateVisibility.NeverReturnExpired – 过期的数据不会返回
- StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanUp – 如果还未被清除则返回过期数据
当使用 NeverReturnExpired 时，过期state的表现就像是真的不存在了一样，即便它又可能还未被完全移除。这个策略在数据过去后就不可见的场景中很适合，例如与敏感隐私数据交互的应用。
另一个选项 ReturnExpiredIfNotCleanUp 允许返回未被清除的过期数据。
注意：
- state数据进行最后修改的时间戳会和state一起存储在状态后端，也意味着会增加状态存储的消耗。使用堆内存作为状态后端时，会使用另一个类，该类拥有指向用户state对象的引用以及一个private long值。使用RocksDB作为状态后端时，会在每个存储的value，每个list的元素以及每个map的entry前添加8字节。
- 当前仅支持 TTL 的时间引用 processing time
- 使用TTL重新存储之前为配置TTL时的state，会导致兼容故障与 StateMigrationException，反之亦然。
- TTL 配置不是checkpoint/savepoint 的一部分，而是一种说明flink如何处理当前正在运行的job。
- 使用TTL配置的map state，如果user value的序列化器可以处理null值，则允许其写入null。如果序列化器不支持null值，它可以被 NullableSerializer 包装起来，但是会在序列化框架中多耗费额外的字节。
Cleanup of Expired State
默认情况下，过期的数据只有在读取超时时，被清除，也就是调用 ValueState.value()。
注意：这意味着，默认情况下，如果过期state没有被read，则不会被清除，可能会导致state不断增长。这可能会在以后的版本中解决。
在snapshot中进行cleanup
除了上面的默认方式外，你也可以在执行 full state snapshot时触发cleanup，这样也会减少state snapshot 的大小。在目前的实现中，本地state不会被清除，但是当从以前的snapshot恢复时，不会加载snapshot中过期的state。这种方式可以在 StateTtlConfig中配置。
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
.cleanupFullSnapshot()
.build();
这种方式不合适RocksDB状态后端的incremental checkpoint。
注意：
- 对于已经存在的job来说，clearup 策略可以随时使用 StateTtlConfig 来生效或者失效，如：修改配置后，从savepoint处重启
Incremental cleanup
另一个选项是，当一些state有增量变化时，触发cleanup。可以在每次访问state或者处理每条数据时的回调函数中触发cleanup。如果对某些state使用了这种cleanup策略，存储后端就会维持一个包含所有state值的iterator。每次触发cleanup时，都会迭代一遍，被扫描的state值都会被检查，过期的值就会被清除。
这个策略可以在 StateTtlConfig中配置：
import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
.cleanupIncrementally()
.build();
这个策略有两个参数。第一个参数为数字，代表每次触发cleanup时需要检查的state的值的数量。如果启用了这个值，每次state被访问时，都会触发cleanup。第二个参数定义了是否需要在每次数据处理是调用cleanup。
注意：
- 如果该state被访问或者没有数据被处理，过期的数据不会被清除
- 为了cleanup所花费的时间，会导致处理延迟的增加
- 目前，incremental cleanup的实现仅针对 Heap 状态后端。如果设置的状态后端为 RocksDB，则不会生效
- 如果Heap状态后端使用同步snapshot机制，全局iterator会在迭代时，维持所有key的副本，因为目前它不支持并发修改，因此这会增加内存消耗。异步snapshot不会有这个问题
- 对于已经在运行的job，可以通过修改 StateTtlConfig来激活/失效cleanup 策略，然后通过savepoint重启。
Cleanup during RocksDB compaction
如果使用RocksDB作为状态后端，另一个cleanup策略是启用Flink的 compaction filter。RocksDB周期性的异步的 压缩合并的state减少存储。Flink compaction filter会检查state的过期数据，这些数据会被排除。
这个特性默认是关闭的。首先需要配置状态后端为RocksDB，然后可以通过设置 state.backend.rocksdb.ttl.compaction.filter.enabled或者调用RocksDBStateBackend::enableTtlCompactionFilter。这样，配置使用TTL的state就可以使用该策略清除数据了。
import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
.cleanupInRocksdbCompactFilter()
.build();
RocksDB compaction filter会在处理完特定数量的state值后，从flink查询当前timestamp用于检查过期。这个数值默认是100.你可以通过 StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries) 来修改。经常更新timestamp可以提高cleanup的速度，但是会掉那个地compaction的性能，因为compaction使用JNI来调用native code。
你可以配置 FlinkCompactionFilter 来启动 RocksDB filter 的debug 日志：
log4j.logger.org.rocksdb.FlinkCompactionFilter=DEBUG
注意：
- 在 compaction 是调用 TTL filter会降低它的速度。TTL filter要解析最后一个访问state的时间戳，然后在正在 compact 的数据中判断每个数据是否过期。如果state 的类型是集合（如list或map），那么就是检查集合的每个元素。
- 如果这个策略使用在list state中，且list中的元素的字节长度不固定，原生的TTL filter会针对集合的元素，额外的调用flink的java type serializer，来处理集合中每个元素，至少是第一个过期的元素，以便决定下一个过期元素的offset。
- 对于已经在运行的job，可以通过修改 StateTtlConfig来激活/失效cleanup 策略，然后通过savepoint重启。
State in the Scala DataStream API
scala的api暂不翻译
使用 Managed Operator 状态
为了使用 managed operator state，一个有状态的function必须实现 CheckpointedFunction 接口或
ListCheckpointed<T extends Serializable> 接口。
CheckpointedFunction
CheckpointedFunction 接口提供了访问 non-keyed state 以及 不同重分配方案的能力。他要求实现两个方法：
void snapshotState(FunctionSnapshotContext context) throws Exception;
void initializeState(FunctionInitializationContext context) throws Exception;
一旦checkpoint被执行，就会调用 snapshotState() 方法。与之相对的，initializeState() 方法在每次用户自定义函数初始化时被调用，可以使第一次初始化时，也可以是从某个checkpoint故障恢复时。基于此，initializeState() 方法不仅是各种state初始化的地方，也是state从故障处恢复的地方。
目前，managed operator state支持list-style 的值。List的元素需要是一个可序列化的对象，且相互之间独立，这样才可以在应用伸缩时进行state的重分配。换句话说，这些对象是在重新分配non-key state时的最小粒度。取决于状态的访问方法，定义了下面两个重分配方案：
- Even-split redistribution:每个operator都有会一个含有state数值的list。逻辑上讲，所有的state是所有list的合并。当重启或者重分配时，list会被均匀的划分到每个并发操作符实例中。每一个操作符都会有一个sublist，它可以是空，也可以有一个或多个元素。如，在并发度为1时，操作符包含两个元素 e1,e2，当增加并发度到2时，可能e1在第一个操作符实例，e2在第二个操作符实例中。
- Union redistribution:每个operator都有会一个含有state数值的list。逻辑上讲，所有的state是所有list的合并。当重启或者重分配时，每个操作符都持有含有所有的state数值的list。
下面的示例展示了一个实现了 CheckpointedFunction 的 SinkFunction ，它会在将数据写入其他系统时，缓存数据。它展示了 even-split redistribution ：
public class BufferingSink
implements SinkFunction<Tuple2<String, Integer>>,
CheckpointedFunction {
private final int threshold;
private transient ListState<Tuple2<String, Integer>> checkpointedState;
private List<Tuple2<String, Integer>> bufferedElements;
public BufferingSink(int threshold) {
this.threshold = threshold;
this.bufferedElements = new ArrayList<>();
}
@Override
public void invoke(Tuple2<String, Integer> value, Context contex) throws Exception {
bufferedElements.add(value);
if (bufferedElements.size() == threshold) {
for (Tuple2<String, Integer> element: bufferedElements) {
// send it to the sink
}
bufferedElements.clear();
}
}
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
checkpointedState.clear();
for (Tuple2<String, Integer> element : bufferedElements) {
checkpointedState.add(element);
}
}
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
ListStateDescriptor<Tuple2<String, Integer>> descriptor =
new ListStateDescriptor<>(
"buffered-elements",
TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));
checkpointedState = context.getOperatorStateStore().getListState(descriptor);
if (context.isRestored()) {
for (Tuple2<String, Integer> element : checkpointedState.get()) {
bufferedElements.add(element);
}
}
}
}
initializeState 的入参为 FunctionInitializationContext。它被用来初始化 non-keyed state。上面的示例中，使用了 ListState 类型的state，用来存储需要被checkpoint的数据。
注意state是如何初始化的，同 keyed state类似，使用 StateDescriptor来获取初始化的state。
ListStateDescriptor<Tuple2<String, Integer>> descriptor =
new ListStateDescriptor<>(
"buffered-elements",
TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}));
checkpointedState = context.getOperatorStateStore().getListState(descriptor);
获取state的方法的名称代表了该state所使用的重分配策略。例如，想要对list state使用 union redistribution策略，则需要在获取state时调用 getUnionListState(descriptor)方法。如果方法名不包含重分配策略的信息，如：getListState(descriptor)，那么，它默认实现的是 even-split redistribution 策略来重分配。
初始化 state 后，我们使用context的 isRestored() 方法来判断是否是故障恢复的情况。如果是的话，就需要在初始化时，应用我们自定义的故障恢复逻辑。
就像在 BufferingSink 中演示的那样，在状态初始化期间生成的 ListState 会保存在类变量中，以供将来可以在 snapshotState() 方法中使用。snapshotState() 方法中，先调用 ListState.clear() 方法将上次checkpoint的数据清空，再填上这次checkpoint需要的数据。
作为旁注：keyed state也可以在 initializeState() 方法中初始化。可以使用提供的 FunctionInitializationContext 来完成。
ListCheckpointed
ListCheckpointed 是一个在 CheckpointedFunction 基础上加了限制的变种，它仅支持list-style 状态使用 even-split 重分配策略。它也要求实现下面的两个方法：
List<T> snapshotState(long checkpointId, long timestamp) throws Exception;
void restoreState(List<T> state) throws Exception;
snapshotState() 方法应该返回需要被checkpoint的方法，而restoreState 会处理一个类似的list用于故障恢复。如果state不是 re-partitionable 的，你可以在snapshotState()返回 Collections.singletonList(MY_STATE).
Stateful Source Functions
stateful source与其他操作符相比，需要更多的操作。为了使状态的更新与数据的发射变得原子性（要求在故障/恢复时做到exactly-once语义），用户需要从source context中获取一个所。
public static class CounterSource
extends RichParallelSourceFunction<Long>
implements ListCheckpointed<Long> {
/** current offset for exactly once semantics */
private Long offset = 0L;
/** flag for job cancellation */
private volatile boolean isRunning = true;
@Override
public void run(SourceContext<Long> ctx) {
final Object lock = ctx.getCheckpointLock();
while (isRunning) {
// output and state update are atomic
synchronized (lock) {
ctx.collect(offset);
offset += 1;
}
}
}
@Override
public void cancel() {
isRunning = false;
}
@Override
public List<Long> snapshotState(long checkpointId, long checkpointTimestamp) {
return Collections.singletonList(offset);
}
@Override
public void restoreState(List<Long> state) {
for (Long s : state)
offset = s;
}
}
某些operator可能需要知道Flink是什么时候确认checkpoint完成的，然后将这个信息与外部系统交互。这种情况可以参阅 org.apache.flink.runtime.state.CheckpointListener 接口。
