Checkpoint

温故知新：

1. Spark RDD Checkpoint：将RDD的数据保存到可靠文件系统，比如HDFS。
2. Flink Checkpoint：将某一时刻的状态State数据（状态快照）保存到可靠文件系统，比如HDFS。用途：用于将任务恢复到指定的状态。

State：存储的是某一个Operator或Key的运行的状态/历史值，默认维护在TaskManager节点的Java堆内存内存中。

Checkpoint是Flink实现容错机制最核心的功能，根据配置周期性地基于Stream中各个Operator/task的状态State来生成快照，从而将这些状态数据定期持久化存储下来，当Flink程序一旦意外崩溃或集群扩容时，重新运行程序时可以有选择地从这些快照进行恢复，从而修正因为故障带来的程序数据异常。

Flink会在输入的数据集上间隔性地生成checkpoint barrier，通过栅栏（barrier）将间隔时间段内的数据划分到相应的checkpoint中。

1. Flink的JobManager创建CheckpointCoordinator；
2. CheckpointCoordinator向所有的SourceOperator发送Barrier栅栏(理解为执行Checkpoint的信号)；
3. SourceOperator接收到Barrier之后，暂停当前的操作(暂停的时间很短，因为后续的写快照是异步的)，并制作State快照, 然后将自己的快照保存到指定的介质中(如HDFS), 一切 ok之后向Coordinator汇报并将Barrier发送给下游的其他Operator；
4. 下游的TransformationOperator(filter、flatMap、map、keyBy、sum)接收到Barrier，依次重复第3步，最后将Barrier发送给Sink；
5. Sink接收到Barrier之后重复第3步；
6. CheckpointCoordinator接收到所有的Operator和Sink的异步快照存储执行完成的确认汇报结果，认为本次快照执行成功。

栅栏对齐（Barrier Alignment）：下游subTask必须接收到上游的所有SubTask发送Barrier栅栏信号，才开始进行Checkpoint操作。

扩展：Flink的Checkpoint的核心算法叫做Chandy-Lamport