Flink增量式checkpoint介绍
在Flink中管理⼤量的状态--增量式的检查点的介绍
本⽂由 翻译⽽来，为了叙述的可读性和流畅性，笔者做了少量的修改。
Apache Flink是为了“有状态”的处理流式数据建⽴的。那么，在流式计算程序中，状态的含义是什么？ 我在前⾯的中做了“状态”以及“有状态的流式处理”的定义。这⾥回顾⼀下，状态指的是，在程序中，Operator将过去处理过的event信息保存在内存中， 这样可以在之后的处理中使⽤。
“状态”是⼀个基础的功能，使得在流式计算中复杂的⽤户使⽤场景成为可能。在中列举 了⼀些例⼦。
程序需要查找某些固定的模式的事件，“状态”保存了⾄今接收到的事件流。三年级科学上册
程序需要每分钟做聚合操作，“状态”缓存等待聚合的数据
程序需要基于流式的数据进⾏模型训练，“状态”保存当前版本的模型参数。
但是，只有“状态”拥有容错能⼒，这样才能在⽣产环境使⽤。“容错性”意味着，即使有软件或者机器的故障，最终的计算结果也是精确的，没有数据丢失也没有重复处理。
Flink的容错特性⾮常强⼤，它不仅对软件和机器的负载很⼩，并且也提供了“端到端仅⼀次”的消息传递保证。
Flink程序容错机制的核⼼是检查点。Flink的检查点是⼀个全局的、异步的程序快照，它周期性的⽣成并送到持久化存储（⼀般使⽤分布式系统）。 当发⽣故障时，Flink使⽤最新的检查点进⾏重启。⼀些Flink的⽤户在程序“状态”中保存了GB甚⾄TB的数据。这些⽤户反馈在⼤量 的状态下，创建检查点通常很慢并且耗资源，这也是为什么Flink在 1.3版本开始引⼊“增量式的检查点”。
在引⼊“增量式的检查点”之前，每⼀个Flink的检查点都保存了程序完整的状态。后来我们意识到在⼤部分情况下这是不必要的，因为上⼀次和这次的检查点之前 ，状态发⽣了很⼤的变化，所以我们创建了“增量式的检查点”。增量式的检查点仅保存过去和现在状态的差异部分。结球白菜
好久不见了
增量式的检查点可以为拥有⼤量状态的程序带来很⼤的提升。在早期的测试中，⼀个拥有TB级别“状态”程序将⽣成检查点的耗时从3分钟以上降低 到了30秒左右。因为增量式的检查点不需要每次把完整的状态发送到存储中。
如何使⽤
现在只能通过RocksDB state back-end来获取增量式检查点的功能，Flink使⽤RocksDB内置的备份机
制来合并检查点数据。这样，Flink 增量式检查点的数据不会⽆限制的增⼤，它会⾃动合并⽼的检查点数据并清理掉。
想要在程序中使⽤增量式的检查点，我建议详细的阅读 总的来说，要启⽤这个机制，可以如下设置：
StreamExecutionEnvironment env = ExecutionEnvironment(); env.tStateBackend(new RocksDBStateBackend(filebackend, true));//第⼆个参数为true
默认的，Flink保留⼀个完整的检查点，如果你需要保留更多，可以:
state.checkpoints.num-retained
它是怎么⼯作的
Flink 增量式的检查点以“”为基础，RocksDB是⼀个基于 LSM树的KV存储，新的数据保存在内存中，称为memtable。如果Key相同，后到的数据将覆盖之前的数据，⼀旦memtable写满了，RocksDB将数据压缩并写⼊到磁盘。memtable的数据持久化到磁盘后，他们就变成了不可变的sstable。
RocksDB会在后台执⾏compaction，合并sstable并删除其中重复的数据。之后RocksDB删除原来的sstable，替换成新合成的ssttable，这个sstable包含了之前的sstable中的信息。
在这个基础之上，Flink跟踪前⼀个checkpoint创建和删除的RocksDB sstable⽂件，因为sstable是不可变的，Flink可以因此计算出 状态有哪些改变。为了达到这个⽬标，Flink在RocksDB上触发了⼀个刷新操作，强制将memtable刷新到磁盘上。这个操作在Flink中是同步的，其他的操作是异步的，不会阻塞数据处理。
Flink 的checkpoint会将新的sstable发送到持久化存储（例如HDFS，S3）中，同时保留引⽤。Flink不会发送所有的sstable， ⼀些数据在之前的checkpoint存在并且写⼊到持久化存储中了，这样只需要增加引⽤次数就可以了。因为compaction的作⽤，⼀些sstable会合并成⼀个sstable并删除这些sstable，这也是为什么Flink可以减少checkpoint的历史⽂件。
太阳系的八大行星
为了分析checkpoint的数据变更，⽽上传整理过的sstable是多余的（下⽂会有描述，这⾥的意思是之前已经上传过的，不需要再次上传）。Flink处理这种情况，仅带来⼀点点开销。这个过程很重要，因为在任务需要重启的时候，Flink只需要保留较少的历史⽂件。
假设有⼀个⼦任务，拥有⼀个keyed state的operator，checkpoint最多保留2个。上⾯的图⽚描述了每个checkpoint对应的RocksDB 的状态，它引⽤到的⽂件，以及在checkpoint完成后共享状态中的count值。
仓储能力
checkpoint ‘CP2’，本地的RocksDB⽬录有两个sstable⽂件，这些⽂件是新⽣成的，于是Flink将它们传到了checkpoint 对应的存储⽬录。当checkpoint完成后，Flink在共享状态中创建两个实体，并将count设为1。在这个共享状态中，这个key 由operator、subtask，原始的sstable名字组成，value为sstable实际存储⽬录。
checkpoint‘CP2’,RocksDB有2个⽼的sstable⽂件,⼜创建了2个新的sstable⽂件。Flink将这两个新的sstable传到 持久化存储中，然后引⽤他们。当checkpoint完成后，Flink将所有的引⽤的相应计数加1。
哥尔顿checkpoint‘CP3’,RocksDB的compaction将sstable-(1), sstable-(2), sstable-(3) 合并成 sstable-(1,2,3)，然后删除 原始的sstable。这个合并后的⽂件包含了和之前源⽂件⼀样的信息，并且清理掉了重复的部分。sstable-(4)还保留着，然后有⼀个 新⽣成的sstable-(5)。Flink将新的 sstable-(1,2,3)以及 sstable-(5)传到持久化存储中， sstable-(4)仍被‘CP2’引⽤，所以 将计数增加1。现在有了3个checkpoint，'CP1','CP2','CP3'，超过了预设的保留数⽬2，所以CP1被删除。作为删除的⼀部分， CP1对应的⽂件（sstable-(1)、sstable-(2)） 的引⽤计数减1。
checkpoint‘CP4’，RocksDB将sstable-(4), sstable-(5), 新的 sstable-(6) 合并成 sstable-(4,5,6)。Flink将新合并 的 sstable-(4,5,6)发送到持久化存储中，sstable-(1,2,3)、sstable-(4,5,6) 的引⽤计数增加1。由于再次到达了checkpoint的 保留数
⽬，‘CP2’将被删除，‘CP2’对应的⽂件（sstable-(1)、sstable-(2)、sstable(3) ）的引⽤计数减1。由于‘CP2’对应 的⽂件的引⽤计数达到0，这些⽂件将被删除。机智的反义词
竞态条件和并发的checkpoint
由于Flink可以并⾏的执⾏多个checkpoint，有时候前⾯的checkpoint还没有完成，后⾯的新的checkpoint就启动了。因此，在 使⽤增量式的checkpoint的时候，你需要考虑使⽤哪⼀个checkpoint启动。Flink在使⽤checkpoint之前需要checkpoint协调器的确认， 所以不会使⽤那些被删除的checkpoint。
通过checkpoint恢复状态，以及性能的注意事项
如果使⽤增量式的checkpoint，那么在错误恢复的时候，不需要考虑很多的配置项。⼀旦发⽣了错误，Flink的JobManager会告诉 task需要从最新的checkpoint中恢复，它可以是全量的或者是增量的。之后TaskManager从分布式系统中下载checkpoint⽂件， 然后从中恢复状态。
增量式的checkpoint能为拥有⼤量状态的程序带来较⼤的提升，但还有⼀些trade-off需要考虑。总的来说，增量式减少了checkpoint操作的时间，但是相对的，从checkpoint中恢复可能更耗时，具体情况需要根据应⽤程序包含的状态⼤⼩⽽定。相对的，如果程序只是部分失败，Flink TaskManager需要从
多个checkpoint中读取数据，这时候使⽤全量的checkpoint来恢复数据可能更加耗时。同时，由于新的checkpoint可能引⽤到⽼的checkpoint，这样⽼的checkpoint就不能被删除，这样下去，历史的版本数据会越来越⼤。需要考虑使⽤分布式来存储checkpoint，另外还需要考虑读取带来的带宽消耗。供货合同协议
还有⼀些便利性和性能的trade-off，可以通过阅读 了解更多。
