1.概述

Flink支持有状态计算，根据支持得不同状态类型，分别有Keyed State和Operator State。针对状态数据得持久化，Flink提供了Checkpoint机制处理；针对状态数据，Flink提供了不同的状态管理器来管理状态数据，如MemoryStateBackend。

上面Flink的文章中，有引用word count的例子，但是都没有包含状态管理。也就是说，如果一个task在处理过程中挂掉了，那么它在内存中的状态都会丢失，所有的数据都需要重新计算。

从容错和消息处理的语义上(at least once, exactly once)，Flink引入了state和checkpoint。

首先区分一下两个概念，state一般指一个具体的task/operator的状态。而checkpoint则表示了一个Flink Job，在一个特定时刻的一份全局状态快照，即包含了所有task/operator的状态。

所谓checkpoint，就是在某一时刻，将所有task的状态做一个快照(snapshot)，然后存储到memory/file system/rocksdb等。Flink通过定期地做checkpoint来实现容错和恢复。

2.checkpoint的实现CheckpointedFunction

CheckpointedFunction的描述

/** * This is the core interface for <i>stateful transformation functions</i>, meaning functions

* that maintain state across individual stream records.

* While more lightweight interfaces exist as shortcuts for various types of state, this interface offer the

* greatest flexibility in managing both <i>keyed state</i> and <i>operator state</i>.

*

* <p>The section <a href="#shortcuts">Shortcuts</a> illustrates the common lightweight

* ways to setup stateful functions typically used instead of the full fledged

* abstraction represented by this interface.

*

* <h1>Initialization</h1>

* The {@link CheckpointedFunction#initializeState(FunctionInitializationContext)} is called when

* the parallel instance of the transformation function is created during distributed execution.

* The method gives access to the {@link FunctionInitializationContext} which in turn gives access

* to the to the {@link OperatorStateStore} and {@link KeyedStateStore}.

*

* <p>The {@code OperatorStateStore} and {@code KeyedStateStore} give access to the data structures

* in which state should be stored for Flink to transparently manage and checkpoint it, such as

* {@link org.apache.flink.api.common.state.ValueState} or

* {@link org.apache.flink.api.common.state.ListState}.

*

* <p><b>Note:</b> The {@code KeyedStateStore} can only be used when the transformation supports

* <i>keyed state</i>, i.e., when it is applied on a keyed stream (after a {@code keyBy(...)}).

*

* <h1>Snapshot</h1>

* The {@link CheckpointedFunction#snapshotState(FunctionSnapshotContext)} is called whenever a

* checkpoint takes a state snapshot of the transformation function. Inside this method, functions typically

* make sure that the checkpointed data structures (obtained in the initialization phase) are up

* to date for a snapshot to be taken. The given snapshot context gives access to the metadata

* of the checkpoint.

*

* <p>In addition, functions can use this method as a hook to flush/commit/synchronize with

* external systems.

*

* <h1>Example</h1>

* The code example below illustrates how to use this interface for a function that keeps counts

* of events per key and per parallel partition (parallel instance of the transformation function

* during distributed execution).

* The example also changes of parallelism, which affect the count-per-parallel-partition by

* adding up the counters of partitions that get merged on scale-down. Note that this is a

* toy example, but should illustrate the basic skeleton for a stateful function.

*

* <p><pre>{@code * public class MyFunction<T> implements MapFunction<T, T>, CheckpointedFunction {

*

* private ReducingState<Long> countPerKey;

* private ListState<Long> countPerPartition;

*

* private long localCount;

*

* public void initializeState(FunctionInitializationContext context) throws Exception {

* // get the state data structure for the per-key state

* countPerKey = context.getKeyedStateStore().getReducingState(

* new ReducingStateDescriptor<>("perKeyCount", new AddFunction<>(), Long.class));

*

* // get the state data structure for the per-partition state

* countPerPartition = context.getOperatorStateStore().getOperatorState(

* new ListStateDescriptor<>("perPartitionCount", Long.class));

*

* // initialize the "local count variable" based on the operator state

* for (Long l : countPerPartition.get()) {

* localCount += l;

* }

* }

*

* public void snapshotState(FunctionSnapshotContext context) throws Exception {

* // the keyed state is always up to date anyways

* // just bring the per-partition state in shape

* countPerPartition.clear();

* countPerPartition.add(localCount);

* }

*

* public T map(T value) throws Exception {

* // update the states

* countPerKey.add(1L);

* localCount++;

*

* return value;

* }

* }

* }</pre>

*

* <hr>

*

* <h1><a name="shortcuts">Shortcuts</a></h1>

* There are various ways that transformation functions can use state without implementing the

* full-fledged {@code CheckpointedFunction} interface:

*

* <h4>Operator State</h4>

* Checkpointing some state that is part of the function object itself is possible in a simpler way

* by directly implementing the {@link ListCheckpointed} interface.

* That mechanism is similar to the previously used {@link Checkpointed} interface.

*

* <h4>Keyed State</h4>

* Access to keyed state is possible via the {@link RuntimeContext}'s methods:

* <pre>{@code * public class CountPerKeyFunction<T> extends RichMapFunction<T, T> {

*

* private ValueState<Long> count;

*

* public void open(Configuration cfg) throws Exception {

* count = getRuntimeContext().getState(new ValueStateDescriptor<>("myCount", Long.class));

* }

*

* public T map(T value) throws Exception {

* Long current = count.get();

* count.update(current == null ? 1L : current + 1);

*

* return value;

* }

* }

* }</pre>

*

* @see ListCheckpointed

* @see RuntimeContext

*/