Flink中的每个function与operator都可以是有状态的。有状态的function会在处理每个数据时存储数据，使得状态state成为需要更多精细操作的任意类型的operator的必要的构造块。
为了实现状态的故障恢复，Flink需要对状态做快照checkpoint。使用checkpoint允许Flink恢复job的状态与流的位置，给应用提供一个一致的语义，就像从来没有故障发生过。
文档 documentation on streaming fault tolerance 描述了Flink 故障容忍机制背后的技术细节。

Prerequisites 前提

Flink的checkpoint 机制需要为流和state提供稳定的存储。通常，这要求：

• 一个持久化（稳定的）数据源，可以从某个特定时刻回放数据。如持久化数据队列（如 kafka,RabbitMQ,Amazon Kinesis,Google PubSub） 或 文件系统（如：HDFS,S3,GFS,NFS,Ceph）
• 一个存储state的持久化存储，通常是分布式文件系统（如：HDFS,S3,GFS,NFS,Ceph…）

开启与配置 Checkpointing

默认情况下，checkpoint是关闭的。为了开启checkpoint，在 StreamExecutionEnvironment 上调用 enableCheckpointing(n) 方法，其中n代表checkpoint的频率（毫秒）。
checkpoint的其他参数有：

• exactly-once vs. at-least-once:你可以在调用 enableCheckpointing(n)时传入一个可选参数 mode ，仅有两种mode代表Flink提供的两种级别的保证。Exactly-once 对大多数应用都是最好的选择。 at-least-once 可能适合那些需要超低延迟（仅有几毫秒）的应用。
• checkpoint timeout:如果在配置的时间后，仍然没有完成checkpoint ，该次checkpoint会被抛弃。
• minimum time between checkpoints:为了保证在流应用中不会被checkpoint占用太多进程，你可以配置两次checkpoint间的最小时间间隔。如果这个值设置为，如5000，下一次checkpoint最早将会在上次checkpoint执行完毕后的5秒后执行，不管checkpoint花了多久，也不管配置的checkpoint interval。注意，使用这个配置也就意味着checkpoint interval不能比该值小
  在应用中配置“time between checkpoints”比配置checkpoint interval容易的多，因为在某些checkpoint花费比平均时间更长的时间时（如目标存储系统短暂缓慢时），“time between checkpoints”不会受影响
  注意的是这个配置也意味着checkpoint的并发度为1。
• number of current checkpoint:默认情况下，系统不会在上次checkpoint未完成的情况下触发新的checkpoint。这保证了topology拓扑结构不会在checkpoint上花费太多时间。Flink允许多个checkpoint进程同时进行，一个有趣的例子是在pipeline场景中有一个特定的处理延迟（如某个function调用远程服务，需要一些时间等待响应），或者是那些想要很频繁的checkpoint(如百分之一毫秒)来实现遇到故障时快速恢复。
  当定义了”minimum time between checkpoints”时，这个选项不会生效。
• externalized checkpoints:你可以配置checkpoint周期性的持久化到外部系统。externalized checkpoints 会将元信息写入持久化存储系统中，当job故障时也不会被自动清除。deployment notes on externalized checkpoint有更详细的说明。
• fail/continue task on checkpoint errors:这个配置决定当执行task的checkpoint过程遇到异常时，该task是否会fail。默认是fail。当选择另一选项时，task会拒绝checkpoint coodinator 的checkpoint请求，继续运行。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// start a checkpoint every 1000 ms
env.enableCheckpointing(1000);

// advanced options:

// set mode to exactly-once (this is the default)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// make sure 500 ms of progress happen between checkpoints
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

// checkpoints have to complete within one minute, or are discarded
env.getCheckpointConfig().setCheckpointTimeout(60000);

// allow only one checkpoint to be in progress at the same time
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// enable externalized checkpoints which are retained after job cancellation
env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

相关的配置选项

更多的参数或默认参数可以参阅 conf/flink-conf.yaml 文件

选择一个状态后端

Flink的checkpoint机制会存储timer与有状态的operater中的state的快照，有状态的operator包括connector,window以及所有自定义的状态。快照存储到哪里，取决于配置的 状态后端 State Backend。
默认情况下，状态存储在TaskManager的内存中，而快照存储在JobManager的内存中。为了存储大量的state，flink支持多种状态后端用于存储状态以及存储状态快照。可以通过如下方式配置状态后端：
StreamExecutionEnvironment.setStateBackend(...)
查阅 state backends 了解可选的状态后端，以及可对job与cluster做的配置。

Iterative(迭代的)job中的状态快照state checkpoint

flink目前仅对不使用iteration的job提供chckpoint处理保证。允许在iteration job中使用checkpoint会产生异常。为了强制在iterative 程序中使用checkpoint，使用者需要在开启checkpoint时设置一个特殊的标志：
env.enableCheckpointing(interval,CheckpointingMode.EXACTLY_ONCE,force = true)

重启策略

Flink提供了多种不同的重启策略，详情参阅Restart Strategies