Flink具体如何保证exactly-once呢? 它使用一种被称为"检查点"（checkpoint）的特性，在出现故障时将系统重置回正确状态

1 Flink的检查点算法

Flink检查点的核心作用是确保状态正确，即使遇到程序中断，也要正确。记住这一基本点之后，我们用一个例子来看检查点是如何运行的。Flink为用户提供了用来定义状态的工具。例如，以下这个Scala程序按照输入记录的第一个字段(一个字符串)进行分组并维护第二个字段的计数状态。

val stream: DataStream[(String, Int)] = ...
val counts: DataStream[(String, Int)] = stream.keyBy(record => record._1).mapWithState(  (in: (String, Int), state: Option[Int])  => state match { case Some(c) => ( (in._1, c + in._2), Some(c + in._2) ) case None => ( (in._1, in._2), Some(in._2) )
})

该程序有两个算子: keyBy算子用来将记录按照第一个元素(一个字符串)进行分组，根据该key将数据进行重新分区，然后将记录再发送给下一个算子: 有状态的map算子(mapWithState)。map算子在接收到每个元素后，将输入记录的第二个字段的数据加到现有总数中，再将更新过的元素发射出去。

下图表示程序的初始状态: 输入流中的6条记录被检查点分割线(checkpoint barrier)隔开，所有的map算子状态均为0(计数还未开始)。所有key为a的记录将被顶层的map算子处理，所有key为b的记录将被中间层的map算子处理，所有key为c的记录则将被底层的map算子处理。

上图是程序的初始状态。注意，a、b、c三组的初始计数状态都是0，即三个圆柱上的值。ckpt表示检查点分割线（checkpoint barriers）。每条记录在处理顺序上严格地遵守在检查点之前或之后的规定，例如["b",2]在检查点之前被处理，["a",2]则在检查点之后被处理。

当该程序处理输入流中的6条记录时，涉及的操作遍布3个并行实例(节点、CPU内核等)。那么，检查点该如何保证exactly-once呢?

检查点分割线和普通数据记录类似。它们由算子处理，但并不参与计算，而是会触发与检查点相关的行为。当读取输入流的数据源(在本例中与keyBy算子内联)遇到检查点屏障时，它将其在输入流中的位置保存到持久化存储中。如果输入流来自消息传输系统(Kafka)，这个位置就是偏移量。Flink的存储机制是插件化的，持久化存储可以是分布式文件系统，如HDFS。下图展示了这个过程。

当Flink数据源(在本例中与keyBy算子内联)遇到检查点分界线（barrier）时，它会将其在输入流中的位置保存到持久化存储中。这让 Flink可以根据该位置重启。

检查点像普通数据记录一样在算子之间流动。当map算子处理完前3条数据并收到检查点分界线时，它们会将状态以异步的方式写入持久化存储，如下图所示：

位于检查点之前的所有记录(["b",2]、["b",5]和["c",1])被map算子处理之后的情况。此时，持久化存储已经备份了检查点分界线在输入流中的位置(备份操作发生在barrier被输入算子处理的时候)。map算子接着开始处理检查点分界线，并触发将状态异步备份到稳定存储中这个动作。

当map算子的状态备份和检查点分界线的位置备份被确认之后，该检查点操作就可以被标记为完成，如下图所示。我们在无须停止或者阻断计算的条件下，在一个逻辑时间点(对应检查点屏障在输入流中的位置)为计算状态拍了快照。通过确保备份的状态和位置指向同一个逻辑时间点，后文将解释如何基于备份恢复计算，从而保证exactly-once。值得注意的是，当没有出现故障时，Flink检查点的开销极小，检查点操作的速度由持久化存储的可用带宽决定

检查点操作完成，状态和位置均已备份到稳定存储中。输入流中的所有数据记录都已处理完成。值得注意的是，备份的状态值与实际的状态值是不同的。备份反映的是检查点的状态。

如果检查点操作失败，Flink可以丢弃该检查点并继续正常执行，因为之后的某一个检查点可能会成功。虽然恢复时间可能更长，但是对于状态的保证依旧很有力。只有在一系列连续的检查点操作失败之后，Flink才会抛出错误，因为这通常预示着发生了严重且持久的错误。

现在来看看下图所示的情况: 检查点操作已经完成，但故障紧随其后

在这种情况下，Flink会重新拓扑(可能会获取新的执行资源)，将输入流倒回到上一个检查点，然后恢复状态值并从该处开始继续计算。在本例中，["a",2]、["a",2]和["c",2]这几条记录将被重播。

Flink将输入流倒回到上一个检查点屏障的位置，同时恢复map算子的状态值。然后，Flink从此处开始重新处理。这样做保证了在记录被处理之后，map算子的状态值与没有发生故障时的一致。

Flink检查点算法的正式名称是异步分界线快照(asynchronous barrier snapshotting)。该算法大致基于Chandy-Lamport分布式快照算法。

检查点是Flink最有价值的创新之一，因为它使Flink可以保证exactly-once，并且不需要牺牲性能。

Flink的检查点（checkpoint）相关推荐

2021年大数据Flink（二十七）：Flink 容错机制 Checkpoint
目录 Flink 容错机制 Checkpoint State Vs Checkpoint Checkpoint执行流程简单流程复杂流程 State状态后端/State存储介质 MemStateBa ...
【Flink】Flink 清理过期 Checkpoint 目录的正确姿势
1.概述转载:Flink 清理过期 Checkpoint 目录的正确姿势参考:Flink如何清除过期的CheckPoint
Flink 清理过期 Checkpoint 目录的正确姿势
本博客是笔者在生产环境使用 Flink 遇到的 Checkpoint 相关故障后,整理输出,价值较高的实战采坑记,本文会带你更深入的了解 Flink 实现增量 Checkpoint 的细节. 通过本 ...
【flink】95-260-045-源码-检查点-CheckPoint
文章目录 2.概述 2.1 概念 2.2 特点 2.3 简介 3.状态保留策略 4.检查点触发机制 4. Flink Checkpoint 语义 5.Checkpoint的简单设置 5.1 通过配置文 ...
[Flink 日记]Exceeded checkpoint tolerable failure threshold.
org.apache.flink.util.FlinkRuntimeException: Exceeded checkpoint tolerable failure threshold. 原因 Che ...
Mysql检查点CheckPoint、LSN、redo日志的刷新关系
官方文档:https://dev.mysql.com/doc/refman/5.7/en/innodb-checkpoints.html 检查点:为了刷新buffer pool的数据page到磁盘中, ...
【数据库管理】⑧检查点checkpoint
1. checkpoint的本质 Checkpoint是Oracle数据库中的一种机制,用于将内存中的数据写入磁盘,以保证数据的持久性和一致性.Checkpoint的本质是将数据库中的脏数据(即已经被 ...
flink 8 state checkpoint savepoint区别
state 记录各个流程中的状态,比如valuState listState checkpoint 在程序挂掉时,可以从checkpoint恢复任务,程序内部恢复时用 savepoint 在程序kil ...
Flink 利用 Checkpoint 实现故障恢复
传送门:Flink 系统性学习笔记 Flink 1.13.5 版本在本节中,我们将介绍 Flink 如何利用检查点 Checkpoint 实现故障恢复. 1. 配置如果我们的任务已经执行很长时间, ...

Flink的检查点（checkpoint）

1 Flink的检查点算法

Flink的检查点（checkpoint）相关推荐

最新文章

热门文章