图解Flink的Checkpoint机制.pdf

《图解Flink的Checkpoint机制.pdf》由会员分享，可在线阅读，更多相关《图解Flink的Checkpoint机制.pdf（10页珍藏版）》请在淘文阁 - 分享文档赚钱的网站上搜索。

1、图解 Flink 的 Checkpoint 机制 Flink 是一个分布式的流处理引擎，而流处理的其中一个特点就是 7X24。那么，如何保障 Flink 作业的持续运行呢？Flink 的内部会将应用状态(state)存储到本地内存或者嵌入式的 kv 数据库(RocksDB)中，由于采用的是分布式架构，Flink 需要对本地生成的状态进行持久化存储，以避免因应用或者节点机器故障等原因导致数据的丢失，Flink 是通过 checkpoint(检查点)的方式将状态写入到远程的持久化存储，从而就可以实现不同语义的结果保障。通过本文，你可以了解到什么是全局一致性检查点，Flink 内部如何通过检查点实现

2、 Exactly Once 的结果保障。什么是 Checkpoint(检查点)为了保证 state 容错，Flink 提供了处理故障的措施，这种措施称之为checkpoint(一致性检查点)。checkpoint 是 Flink 实现容错的核心功能，主要是周期性地触发 checkpoint，将 state 生成快照持久化到外部存储系统(比如HDFS)。这样一来，如果 Flink 程序出现故障，那么就可以从上一次checkpoint 中进行状态恢复，从而提供容错保障。另外，通过 checkpoint 机制，Flink 可以实现 Exactly-once 语义(Flink 内部的 Exactly-

3、once,关于端到端的exactly_once,Flink 是通过两阶段提交协议实现的)。下面将会详细分析 Flink 的checkpoint 机制。检查点的生成 如上图，输入流是用户行为数据，包括购买(buy)和加入购物车(cart)两种，每种行为数据都有一个偏移量，统计每种行为的个数。第一步：JobManager checkpoint coordinator 触发 checkpoint。第二步：假设当消费到cart，3这条数据时，触发了 checkpoint。那么此时数据源会把消费的偏移量 3 写入持久化存储。第三步：当写入结束后，source 会将 state handle(状态存储路径

4、)反馈给JobManager 的 checkpoint coordinator。第四步：接着算子 count buy 与 count cart 也会进行同样的步骤 第五步：等所有的算子都完成了上述步骤之后，即当 Checkpoint coordinator 收集齐所有 task 的 state handle，就认为这一次的 Checkpoint 全局完成了，向持久化存储中再备份一个 Checkpoint meta 文件，那么整个 checkpoint也就完成了，如果中间有一个不成功，那么本次 checkpoin 就宣告失败。检查点的恢复 通过上面的分析，或许你已经对Flink 的 checkp

5、oint 有了初步的认识。那么接下来，我们看一下是如何从检查点恢复的。任务失败 重启作业 恢复检查点 继续处理数据 上述过程具体总结如下：第一步：重启作业 第二步：从上一次检查点恢复状态数据 第三步：继续处理新的数据 Flink 内部 Exactly-Once 实现 Flink 提供了精确一次的处理语义，精确一次的处理语义可以理解为：数据可能会重复计算，但是结果状态只有一个。Flink 通过 Checkpoint 机制实现了精确一次的处理语义，Flink 在触发 Checkpoint 时会向 Source 端插入 checkpoint barrier，checkpoint barriers 是

6、从 source 端插入的，并且会向下游算子进行传递。checkpoint barriers 携带一个 checkpoint ID，用于标识属于哪一个checkpoint，checkpoint barriers 将流逻辑是哪个分为了两部分。对于双流的情况，通过 barrier 对齐的方式实现精确一次的处理语义。关于什么是 checkpoint barrier，可以看一下 CheckpointBarrier 类的源码描述，如下：/*Checkpoint barriers 用来在数据流中实现 checkpoint 对齐的.*Checkpoint barrier由 JobManager 的 chec

7、kpoint coordinator 插入到 Source 中,*Source 会把 barrier 广播发送到下游算子,当一个算子接收到了其中一个输入流的Checkpoint barrier 时,*它就会知道已经处理完了本次 checkpoint 与上次 checkpoint 之间的数据.*一旦某个算子接收到了所有输入流的 checkpoint barrier 时，*意味着该算子的已经处理完了截止到当前 checkpoint 的数据，*可以触发 checkpoint，并将 barrier 向下游传递 *根据用户选择的处理语义，在 checkpoint 完成之前会缓存后一次 checkpoin

8、t 的数据，*直到本次 checkpoint 完成(exactly once)*checkpoint barrier 的 id 是严格单调递增的 */public class CheckpointBarrier extends RuntimeEvent.可以看出 checkpoint barrier 主要功能是实现 checkpoint 对齐的，从而可以实现 Exactly-Once 处理语义。下面将会对 checkpoint 过程进行分解，具体如下：图 1，包括两个流，每个任务都会消费一条用户行为数据(包括购买(buy)和加购(cart)，数字代表该数据的偏移量，count buy 任务统计

9、购买行为的个数，coun cart 统计加购行为的个数。图 2，触发 checkpoint，JobManager 会向每个数据源发送一个新的checkpoint 编号，以此来启动检查点生成流程。图 3，当 Source 任务收到消息后，会停止发出数据，然后利用状态后端触发生成本地状态检查点，并把该 checkpoint barrier 以及 checkpoint id广播至所有传出的数据流分区。状态后端会在 checkpoint 完成之后通知任务，随后任务会向 Job Manager 发送确认消息。在将 checkpoint barrier 发出之后，Source 任务恢复正常工作。图 4，S

10、ource 任务发出的 checkpoint barrier 会发送到与之相连的下游算子任务，当任务收到一个新的 checkpoint barrier 时，会继续等待其他输入分区的 checkpoint barrier 到来，这个过程称之为 barrier 对齐，checkpoint barrier 到来之前会把到来的数据线缓存起来。图 5，任务收齐了全部输入分区的 checkpoint barrier 之后，会通知状态后端开始生成 checkpoint，同时会把 checkpoint barrier 广播至下游算子。图 6，任务在发出 checkpoint barrier 之后，开始处理因

11、barrier 对齐产生的缓存数据，在缓存的数据处理完之后，就会继续处理输入流数据。图 7，最终 checkpoint barrier 会被传送到 sink 端，sink 任务接收到checkpoint barrier 之后，会向其他算子任务一样，将自身的状态写入checkpoint，之后向 Job Manager 发送确认消息。Job Manager 接收到所有任务返回的确认消息之后，就会将此次检查点标记为完成。使用案例 StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();/c

12、heckpoint 的时间间隔，如果状态比较大，可以适当调大该值env.enableCheckpointing(1000);/配置处理语义，默认是 exactly-onceenv.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);/两个 checkpoint 之间的最小时间间隔，防止因 checkpoint 时间过长，导致 checkpoint 积压env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);/checkpoint 执

13、行的上限时间，如果超过该阈值，则会中断checkpointenv.getCheckpointConfig().setCheckpointTimeout(60000);/最大并行执行的检查点数量，默认为 1，可以指定多个，从而同时出发多个 checkpoint，提升效率 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);/设定周期性外部检查点，将状态数据持久化到外部系统中，/使用该方式不会在任务正常停止的过程中清理掉检查点数据env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);总结 本文首先从 Flink 的状态入手，以图解加文字的形式详细解释了 Flink 的checkpoint 机制，并给出了使用 Checkpoint 时的程序配置。