Flink是一个分布式的流处理引擎,而流处理的其中一个特点就是7X24。那么,如何保障Flink作业的持续运行呢?Flink的内部会将应用状态(state)存储到本地内存或者嵌入式的kv数据库(RocksDB)中,由于采用的是分布式架构,Flink需要对本地生成的状态进行持久化存储,以避免因应用或者节点机器故障等原因导致数据的丢失,Flink是通过checkpoint(检查点)的方式将状态写入到远程的持久化存储,从而就可以实现不同语义的结果保障。通过本文,你可以了解到什么是全局一致性检查点,Flink内部如何通过检查点实现Exactly Once的结果保障。
为了保证state容错,Flink提供了处理故障的措施,这种措施称之为checkpoint(一致性检查点)。checkpoint是Flink实现容错的核心功能,主要是周期性地触发checkpoint,将state生成快照持久化到外部存储系统(比如HDFS)。这样一来,如果Flink程序出现故障,那么就可以从上一次checkpoint中进行状态恢复,从而提供容错保障。另外,通过checkpoint机制,Flink可以实现Exactly-once语义(Flink内部的Exactly-once,关于端到端的exactly_once,Flink是通过两阶段提交协议实现的)。下面将会详细分析Flink的checkpoint机制。
如上图,输入流是用户行为数据,包括购买(buy)和加入购物车(cart)两种,每种行为数据都有一个偏移量,统计每种行为的个数。
第一步:JobManager checkpoint coordinator 触发checkpoint。
第二步:假设当消费到[cart,3]这条数据时,触发了checkpoint。那么此时数据源会把消费的偏移量3写入持久化存储。
第三步:当写入结束后,source会将state handle(状态存储路径)反馈给JobManager的checkpoint coordinator。
第四步:接着算子count buy与count cart也会进行同样的步骤
第五步:等所有的算子都完成了上述步骤之后,即当 Checkpoint coordinator 收集齐所有 task 的 state handle,就认为这一次的 Checkpoint 全局完成了,向持久化存储中再备份一个 Checkpoint meta 文件,那么整个checkpoint也就完成了,如果中间有一个不成功,那么本次checkpoin就宣告失败。
通过上面的分析,或许你已经对Flink的checkpoint有了初步的认识。那么接下来,我们看一下是如何从检查点恢复的。
任务失败
重启作业
恢复检查点
继续处理数据
上述过程具体总结如下:
第一步:重启作业
第二步:从上一次检查点恢复状态数据
第三步:继续处理新的数据
Flink提供了精确一次的处理语义,精确一次的处理语义可以理解为:数据可能会重复计算,但是结果状态只有一个。Flink通过Checkpoint机制实现了精确一次的处理语义,Flink在触发Checkpoint时会向Source端插入checkpoint barrier,checkpoint barriers是从source端插入的,并且会向下游算子进行传递。checkpoint barriers携带一个checkpoint ID,用于标识属于哪一个checkpoint,checkpoint barriers将流逻辑是哪个分为了两部分。对于双流的情况,通过barrier对齐的方式实现精确一次的处理语义。
关于什么是checkpoint barrier,可以看一下CheckpointBarrier类的源码描述,如下:
/**
* Checkpoint barriers用来在数据流中实现checkpoint对齐的.
* Checkpoint barrier由JobManager的checkpoint coordinator插入到Source中,
* Source会把barrier广播发送到下游算子,当一个算子接收到了其中一个输入流的Checkpoint barrier时,
* 它就会知道已经处理完了本次checkpoint与上次checkpoint之间的数据.
*
* 一旦某个算子接收到了所有输入流的checkpoint barrier时,
* 意味着该算子的已经处理完了截止到当前checkpoint的数据,
* 可以触发checkpoint,并将barrier向下游传递
*
* 根据用户选择的处理语义,在checkpoint完成之前会缓存后一次checkpoint的数据,
* 直到本次checkpoint完成(exactly once)
*
* checkpoint barrier的id是严格单调递增的
*
*/
publicclassCheckpointBarrierextendsRuntimeEvent{...}
可以看出checkpoint barrier主要功能是实现checkpoint对齐的,从而可以实现Exactly-Once处理语义。
下面将会对checkpoint过程进行分解,具体如下:
图1,包括两个流,每个任务都会消费一条用户行为数据(包括购买(buy)和加购(cart)),数字代表该数据的偏移量,count buy任务统计购买行为的个数,coun cart统计加购行为的个数。
图2,触发checkpoint,JobManager会向每个数据源发送一个新的checkpoint编号,以此来启动检查点生成流程。
图3,当Source任务收到消息后,会停止发出数据,然后利用状态后端触发生成本地状态检查点,并把该checkpoint barrier以及checkpoint id广播至所有传出的数据流分区。状态后端会在checkpoint完成之后通知任务,随后任务会向Job Manager发送确认消息。在将checkpoint barrier发出之后,Source任务恢复正常工作。
图4,Source任务发出的checkpoint barrier会发送到与之相连的下游算子任务,当任务收到一个新的checkpoint barrier时,会继续等待其他输入分区的checkpoint barrier到来,这个过程称之为barrier 对齐,checkpoint barrier到来之前会把到来的数据线缓存起来。
图5,任务收齐了全部输入分区的checkpoint barrier之后,会通知状态后端开始生成checkpoint,同时会把checkpoint barrier广播至下游算子。
图6,任务在发出checkpoint barrier之后,开始处理因barrier对齐产生的缓存数据,在缓存的数据处理完之后,就会继续处理输入流数据。
图7,最终checkpoint barrier会被传送到sink端,sink任务接收到checkpoint barrier之后,会向其他算子任务一样,将自身的状态写入checkpoint,之后向Job Manager发送确认消息。Job Manager接收到所有任务返回的确认消息之后,就会将此次检查点标记为完成。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// checkpoint的时间间隔,如果状态比较大,可以适当调大该值
env.enableCheckpointing(1000);
// 配置处理语义,默认是exactly-once
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 两个checkpoint之间的最小时间间隔,防止因checkpoint时间过长,导致checkpoint积压
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// checkpoint执行的上限时间,如果超过该阈值,则会中断checkpoint
env.getCheckpointConfig().setCheckpointTimeout(60000);
// 最大并行执行的检查点数量,默认为1,可以指定多个,从而同时出发多个checkpoint,提升效率
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// 设定周期性外部检查点,将状态数据持久化到外部系统中,
// 使用该方式不会在任务正常停止的过程中清理掉检查点数据
env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
本文首先从Flink的状态入手,以图解加文字的形式详细解释了Flink的checkpoint机制,并给出了使用Checkpoint时的程序配置。
免责申明:
本文系转载,版权归原作者所有,如若侵权请联系我们进行删除!