在Flink中的每个函数和运算符都是有状态的。在处理过程中可以用状态来存储数据，这样可以利用状态来构建复杂操作。为了让状态容错，Flink需要设置checkpoint状态。Flink程序是通过checkpoint来保证容错，通过checkpoint机制，Flink可恢复作业的状态和计算位置。
checkpoint检查点
前提条件
Flink的checkpoin机制需要与流和状态的持久化存储交互，一般它要求：
- 一个持久化的数据源
- 当Flink程序出现问题时，可以通过checkpoint持久化存储中恢复，然后从出错的地方开始重新消费数据
- 该数据源可以在一定时间内重跑数据，例如：Kafka、RabbitMQ或者文件系统HDFS、S3、…
- 状态的持久存储
- 状态需要永久的保存下来，通常是分布式文件系统（例如：HDFS、S3、GFS、…）
启用和配置检查点
默认情况，Flink是禁用检查点。要启用检查点，调用
// 启用检查点// 单位：毫秒
env.enableCheckpointing(1000);
在启用检查点时，还可以配置检查点的其他参数。
- exactly-one or at-least-once（仅一次或者至少一次）
- 大多数程序都是设置为exactly-once，只有在某些超低延迟的应用（例如：始终要求是毫秒级的应用）
- 通过查看源码，我们看到，Flink默认是 exactly-once
public static final CheckpointingMode DEFAULT_MODE = CheckpointingMode.EXACTLY_ONCE;
- 检查点超过规定的时间就会自动终止
- 检查点之间的最小时间
- 下一个检查点将在上一个检查点完成后5秒钟启动
- 检查点最小间隔时间不会受检查点间隔更容易配置
- 检查点的并发数目。默认情况一个检查点在运行时不会触发另一个检查点，这样可以确保Flink不会花太多时间在checkpoint上，并确保流可以有效进行。
- 可以设置多个重叠的checkpoint，这对容许有一定延迟，并希望较频繁的检查（100ms）来重新处理故障是有用的
- 外部检查点
- 可以将检查点设置为外部持久化，这样检查点的元数据将写入持久存储，并且但作业运行失败是不会自动清理
- 这样可以做双重保险
- 检查点执行发生错误，是否执行任务。
- 默认情况，如果checkpoint失败，任务也将失败
- 即时最近有更多的savepoint可用于恢复，flink依然会选择使用最近一次的checkpoint来进行错误恢复
参考配置：
// --------
// 配置checkpoint
// 启用检查点
env.enableCheckpointing(1000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.getCheckpointConfig().setPreferCheckpointForRecovery(true);选择状态后端存储
Flink的checkpoint机制可以存储计时器和有状态operation的所有快照，包括：连接器、窗口或者用户自定义状态。具体checkpoint存储在哪儿（例如：是JobManager内存、文件系统或者数据库），依赖于状态后端的配置。
默认情况，状态保存在TaskManager的内存中，检查点存储在TM的内存中。为了适当地保存大状态，Flink支持其他的存储。我们可以通过：
StreamExecutionEnvironment.setStateBackend(…)
来指定存储方式
Flink状态管理
状态的应用场景：
- 当应用程序想要按照某种模式搜索某些事件时，状态可以保存迄今所有的事件序列
- 当每分钟/小时/天需要对流数据进行聚合，状态可以保存挂起的聚合
- 当在数据流上训练机器学习模型时，状态可以用来保存某一类参数的版本
- 当需要管理历史数据时，状态允许访问过去历史数据
Flink状态可以保存在堆内、或者是堆外。Flink也可以管理应用程序的状态，必要时也可以溢出到磁盘，如果应用要保持非常大的状态，可以不修改程序逻辑情况下配置状态后端存储。
Flink状态分类
Flink中有两种基本的状态：
- Keyed State
- Operator State
Keyed State
Keyed State通常和key相关，仅仅在KeyedStream的方法和算子中使用。可以把 Keyed State看作是分区，而且每一个key仅出现在一个分区内。逻辑上每个 keyed-state和唯一元组<算子并发实例, key>绑定，由于每个key仅属于算子的一个并发，因此可以简化为<算子, key>
Operator State
对于 Operator State来说，每个Operator State和一个并发实例绑定。Kafka connector是Flink中使用operator state的一个很好的示例。每个Kafka消费者的并发在Operator State中维护一个 topic partition到offset的映射关系。
Operator state在Flink作业的并发改变后，会重新分发状态，分发的策略和keyed stated不一样。
Raw State与Managed State
Keyed Stated和Operator State分别有两种形式：managed 和 raw
Managed State是由Flink运行时管理的数据结构来表示的，例如：内部的Hash Table或者RocksDB。例如：ValueState、ListState等。Flink运行时会对这些状态进行编码并写入Checkpoint。
Raw State则保存在自己的数据结构中。checkpoint的时候，Flink并不知道状态里面具体的内容，仅仅写入一串字节序列到checkpoint中。
所有的DataStream的function都可以使用managed state，但raw state只能在实现算子时使用。由于Flink可以在修改并发时更好的分发状态数据，并且能够更好的管理内存，因为讲义使用 managed state.
使用Managed Keyed State
Managed keyed state接口提供不同类型的状态访问接口，这些状态都作用在当前输入数据的key下。这些状态仅可在KeyedStream上使用，可以通过 stream.keyBy(…)得到KeyedStream。
所有支持的状态类型如下：
- ValueState<T>
- 保存一个可以更新和获取的值，算子接收到的每个key都可能对应一个值
- 可以通过update(T)进行更新，通过value()获取
- ListState<T>
- 保存一个元素的列表，可以往这个列表中追加数据，并在当前列表上检索
- 可以通过 add(T)或者addAll(List<T>)进行追加元素
- 通过get()获取整个列表
- 通过 update(List<T>)覆盖当前列表
- ReducingState<T>
- 保存一个单值，表示添加到状态的所有值的聚合。接口与ListState类似
- AggregatingState<IN, OUT>
- 保存一个单值，表示添加到状态的所有值的聚合
- 与ReducingState相反的是，聚合类型可能与添加到状态的元素类型不同。接口与ListState类似
- FoldingState<T, ACC>（后续将过期）
- 保存一个单值，白搜狐添加到状态的所有值的集合
- 与ReducingState相反的是，聚合类型可能与添加到状态的元素类型不同。接口与ListState类似
- MapState<UK, UV>
- 维护一个映射列表，可以添加键值到状态中，可以获取当前映射的迭代器
- 使用put、putAll添加映射，使用 get检索特定key
注意：
- 这些状态对象仅用于状态交互。状态本身不一定存储在内存中，还有可能保存在磁盘或者其他位置
- 从状态中获取的值取决于输入元素说代表的key，因此，在不同key上调用同一个接口，可能得到不同的值
使用Managed Operator State
可以通过实现 CheckpointedFunction 或者 ListCheckpointed<T extends Serialized>接口来使用Managed Operator State。
CheckpointedFunction接口：
void snapshotState(FunctionSnapshotContext context) throws Exception;void initializeState(FunctionInitializationContext context) throws Exception;
在Flink进行checkpoint时，会调用snapshotstate()，用户自定义函数初始化时会调用 initializeState。初始化包括第一次自定义函数初始化和从之前的 checkpoint 回复。因此，initializeState 中应该也包括状态恢复的逻辑。
Managed Operator State以list的形式存在，这些状态是一个可序列化对象的集合List，彼此独立，方便在改变并发后进行状态的重新分派。换句话说，这些对象是重新分配 non-keyed state的最细粒度。根据状态的不同访问方式，有以下两种分配模式：
- Even-split redistribution
- 每个算子都存储一个列表形式的状态集合，整个状态由所有的列表拼接而成
- 但作业恢复或者重新分配时，整个状态按照算子的并行度均匀分配
- Union redistribution
- 每个算子保存一个列表形式的状态集合，整个状态由所有的列表拼接而成
- 但作业恢复或者重新分配时，每个算子都将获得所有的状态数据
ListCheckpointed接口：
ListCheckpointed接口是CheckpointedFunction接口的精简版，仅支持 even-split redistribution的list state
List<T> snapshotState(long checkpointId, long timestamp) throws Exception;void restoreState(List<T> state) throws Exception;
snapshotState()需要返回一个将写入到checkpoint的对象列表， restoreState则需要处理恢复回来的对象列表。
参考文献：
Flink官方文档：
https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/dev/stream/state/checkpointing.html
https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/ops/state/checkpoints.html
https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/dev/stream/state/state.html