1、检查点（checkpoint）

Flink 故障恢复机制的核心，就是应用状态的一致性检查点。有状态流应用的一致检查点，其实就是所有任务的状态，在某个时间点的一份拷贝（一份快照）；这个时间点，应该是所有任务都恰好处理完一个相同的输入数据的时候。

下图是一个奇偶求和的例子，sum_even表示偶数求和，sum_odd表示奇数求和，数据源是1-9的数字。从图中可以看出，source偏移量为5，表示已经读取了1-5的数字，sum_even=2+4=6，sum_odd=1+3+5=9。此时做一个checkpoint，将状态保存到状态后端。

从检查点恢复状态

继续读取数据，当读取到数字7时，计算奇数和的程序还没来得及加上7就挂了，此时偶数和为12。

第一步重启应用，此时Source、sum_even和sum_odd的状态都被清空。

第二步是从 checkpoint 中读取状态，将状态重置。从检查点重新启动应用程序后，其内部状态与检查点完成时的状态完全相同。

第三步：开始消费并处理检查点到发生故障之间的所有数据（就是把6和7加上）。这种检查点的保存和恢复机制可以为应用程序状态提供“精确一次”（exactlyonce）的一致性，因为所有算子都会保存检查点并恢复其所有状态，这样一来所有的输入流就都会被重置到检查点完成时的位置。

2、检查点算法

• 一种简单的想法
  暂停应用，保存状态到检查点，再重新恢复应用
• Flink 的改进实现
  基于 Chandy-Lamport 算法的分布式快照
  将检查点的保存和数据处理分离开，不暂停整个应用

检查点分界线（Checkpoint Barrier）

Flink 的检查点算法用到了一种称为分界线（barrier）的特殊数据形式，用来把一条流上数据按照不同的检查点分开。分界线之前到来的数据导致的状态更改，都会被包含在当前分界线所属的检查点中；而基于分界线之后的数据导致的所有更改，就会被包含在之后的检查点中。

算法举例说明：

现在是一个有两个输入流的应用程序，用并行的两个 Source 任务来读取。两个Source 读取的都是1-9的数字。

JobManager 会向每个 source 任务发送一条带有新检查点 ID 的消息，通过这种方式来启动检查点。

数据源将它们的状态写入检查点，并发出一个检查点 barrier。状态后端在状态存入检查点之后，会返回通知给 source 任务，source 任务就会向 JobManager 确认检查点完成。

分界线对齐：barrier 向下游传递，sum 任务会等待所有输入分区的 barrier 到达。对于barrier已经到达的分区，继续到达的数据会被缓存；而barrier尚未到达的分区，数据会被正常处理。

当收到所有输入分区的 barrier 时，任务就将其状态保存到状态后端的检查点中，然后将 barrier 继续向下游转发。

向下游转发检查点 barrier 后，任务继续正常的数据处理。

Sink 任务向 JobManager 确认状态保存到 checkpoint 完毕。当所有任务都确认已成功将状态保存到检查点时，检查点就真正完成了。

3、代码中配置checkpoint

    // 开启checkpoint,参数为生成检查点的时间间隔，默认500ms（watermark默认为200ms）
    // 还可以传第二个参数CheckpointingMode:EXACTLY_ONCE/AT_LEAST_ONCE
    env.enableCheckpointing(1000L)

    // checkpoint的配置
    val chkpConfig = env.getCheckpointConfig
    // 设置生成检查点的时间间隔（前一次Checkpoint开始到下一次Checkpoint开始的时间）
    chkpConfig.setCheckpointInterval(1000L)
    // 设置CheckpointingMode:EXACTLY_ONCE/AT_LEAST_ONCE
    chkpConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // 所有任务完成一个checkpoint的超时时间，一般给分钟级别
    chkpConfig.setCheckpointTimeout(60000L)
    // 同时进行的checkpoint数量，默认1，（一个checkpoint指的是barrier从头到尾完成所有算子的状态保存）
    chkpConfig.setMaxConcurrentCheckpoints(2)
    // 前一次Checkpoint结束到下一次Checkpoint开始至少500ms
    /* 假设第一次Checkpoint在20:00:00开始，两次生成检查点的时间间隔为1s（1000ms）
       1.当Checkpoint处理时间为600ms时，原本下次Checkpoint在20:00:01开始，
       因为最小间隔为500ms，所以现在需要延后100ms，600+500=1.1s
       2.当Checkpoint处理时间为400ms时，下次Checkpoint在20:00:01开始，
       因为最小间隔为500ms，400+500=900ms，小于1s，由于不能提前，所以还是20:00:01开始下次Checkpoint
     */
    // setMaxConcurrentCheckpoints(2)与setMinPauseBetweenCheckpoints(500L)冲突，至少间隔500ms就不可能同时发生了
    chkpConfig.setMinPauseBetweenCheckpoints(500L)
    // 是否使用Checkpoint做故障恢复，不管save point
    chkpConfig.setPreferCheckpointForRecovery(true)
    // 能容忍的Checkpoint失败次数
    chkpConfig.setTolerableCheckpointFailureNumber(3)

4、保存点（save points）

Flink 还提供了可以自定义的镜像保存功能，就是保存点（savepoints）。

• 原则上，创建保存点使用的算法与检查点完全相同，因此保存点可以认为就是具有一些额外元数据的检查点。
• Flink不会自动创建保存点，因此用户（或者外部调度程序）必须明确地触发创建操作。
• 保存点是一个强大的功能。除了故障恢复外，保存点可以用于：有计划的手动备份，更新应用程序，版本迁移，暂停和重启应用，等等。

5、重启策略

// 固定时间间隔重启，参数为重启次数，重启时间间隔
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L))

// 参数为重启次数，所有重启使用的时间，两次重启时间间隔
env.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.minutes(5), Time.seconds(10)))