淘先锋技术网

首页 1 2 3 4 5 6 7

一、什么是CheckPoint?

Flink Checkpoint 是一种容错恢复机制。这种机制保证了实时程序运行时,即使突然遇到异常或者机器问题时也能够进行自我恢复。

二、为什么要开启CheckPoint?

实时任务不同于批处理任务,除非用户主动停止,一般会一直运行,运行的过程中可能存在机器故障、网络问题、外界存储问题等等,要想实时任务一直能够稳定运行,实时任务要有自动容错恢复的功能。

而批处理任务在遇到异常情况时,在重新计算一遍即可。实时任务因为会一直运行的特性,如果在从头开始计算,成本会很大,尤其是对于那种运行时间很久的实时任务来说。

实时任务开启 Checkpoint 功能,也能够减少容错恢复的时间。因为每次都是从最新的 Chekpoint 点位开始状态恢复,而不是从程序启动的状态开始恢复。


三、Flink 任务状态是什么?

Flink 任务状态可以理解为实时任务计算过程中,中间产生的数据结果,同时这些计算结果会在后续实时任务处理时,能够继续进行使用。实时任务的状态可以是一个聚合结果值,比如 WordCount 统计的每个单词的数量,也可以是消息流中的明细数据。

Flink 任务状态整体可以划分两种:Operator 状态和 KeyedState。

  • 常见的 Operator 状态,比如 Kafka Topic 每个分区的偏移量。
  • KeyedState 是基于 KeyedStream 来使用的,所以在使用前,你需要对你的流通过 keyby 来进行分区,常见的状态比如有 MapState、ListState、ValueState 等等。

Flink 整体框架图

如下图所示:
在这里插入图片描述

四、Flink Checkpoint 语义

Flink Checkpoint 支持两种语义:Exactly_Once 和 At_least_Once,默认的 Checkpoint 语义是 Exactly_Once。
具体语义含义如下:

  • Exactly_Once :保证每条数据对于 Flink 任务的状态结果只影响一次。
  • At_Least_Once :每条数据对于 Flink 任务的状态计算至少影响一次。比如在 WordCount 程序中,你统计到的某个单词的单词数可能会比真实的单词数要大,因为同一条消息,当 Flink 任务容错恢复后,可能将其计算多次。

Flink 中 Exactly_Once 和 At_Least_Once 具体是针对 Flink 任务状态而言的,并不是 Flink 程序对消息记录只处理一次。


五、Exactly_Once

目前可以实现exactly_once的不多:

  • source中有kafkaSource
  • sink中有KafkaSink,StreamingFileSink

都要开启checkPoint,并且要保证Source是Exactly_once的,这样才能实现Sink是Exactly_Once

		其中KafkaSource 做ck时,记录偏移量记录在operatorState中,保存在StateBackEnd中,当程
		序出现异常时,根据重启策略重启,从上一次的checkPoint中读取数据,但是会重复读取数据,
		结合Redis、Hbase的幂等性,携带者唯一的ID,会把之前写的数据覆盖掉,从而保证exactly_once。

但是还有一种情况,数据没有唯一的ID,且数据库不支持覆盖,那该怎么保证exactly_once呢?

那存储系统要支持事务,数据提交成功都成功,失败就回滚!


六、以FlinkKafkaProducer为例解析

FlinkKafkaProducer要保证Exactly_once,就要开启checkPoint,还要保证Source是exactly_once的,两者缺一不可。

1、CheckPoint 源码详解

FlinkKafkaProducer继承了TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>这个抽象类,该类又实现了两个接口:

public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>
		extends RichSinkFunction<IN>
		implements CheckpointedFunction, CheckpointListener {

1.1、 CheckpointedFunction中有两个方法:
  • snapshotState:保存偏移量,将结果持久化到StateBackEnd中
  • initializeState:初始化获取ListState(operatorState只支持ListState),或者程序异常重启时,从ListState恢复数据的。
public interface CheckpointedFunction {

	void snapshotState(FunctionSnapshotContext context) throws Exception;

	void initializeState(FunctionInitializationContext context) throws Exception;
}
1.2、 CheckpointListener中有一个方法:
  • notifyCheckpointComplete:当JobManager收到所有的SubTask的checkPoint的ack应答后,通知各个SubTask可以继续工作了。
  • notifyCheckpointComplete中有个commit方法:是让各个SubTask提交事务。
public interface CheckpointListener {

	void notifyCheckpointComplete(long checkpointId) throws Exception;
}

在这里插入图片描述

2、一图总结

2.1、流程图

在这里插入图片描述

2.2、步骤
  • ① JobManager中的CheckPointCadinatior定时器,定时通知各个有状态的SubTask要开启ChenkPoint了
  • ② 建立Barrier(隔离带),隔离数据流,保证一个流水线中的所有算子开始做checkPoint
  • ③ 所有带状态的SubTask将状态数据写入到StateBackEnd中(FsStateBackEnd可以储存在高可用的Hdfs上)
  • ④ CheckPoint成功的SubTask,会向JobManager发起成功的应到
  • ⑤ 当JobManager收齐了所有并行的这些SubTask成功应答后,向所有实现了CheckpointListener(有notifyCheckpointComplete方法)的SubTask发送成功消息,通知它们可以继续完成接下来的工作了。
  • ⑥ 同时所有实现了CheckpointListener的SubTask,会向存储系统(Kafka、MySql)提交事务。

在这里插入图片描述

这里checkPoint和提交事务不能保证同时成功。就算不能同时成功,程序也会重启、也会回滚,这样Flink就保证了数据的Exactly_Once!


七、Flink Checkpoint 常见失败原因分析

Flink Checkpoint 失败有很多种原因,常见的失败原因如下:

  1. 用户代码逻辑没有对于异常处理,让其直接在运行中抛出。比如解析 Json 异常,没有捕获,导致 Checkpoint失败,或者调用 Dubbo 超时异常等等。
  2. 依赖外部存储系统,在进行数据交互时,出错,异常没有处理。比如输出数据到 Kafka、Redis、HBase等,客户端抛出了超时异常,没有进行捕获,Flink 任务容错机制会再次重启。
  3. 内存不足,频繁GC,超出了 GC 负载的限制。比如 OOM 异常
  4. 网络问题、机器不可用问题等等。

从目前的具体实践情况来看,Flink Checkpoint 异常觉大多数还是用户代码逻辑的问题,对于程序异常没有正确的处理导致。所以在编写 Flink 实时任务时,一定要注意处理程序可能出现的各种异常。这样,也会让实时任务的逻辑更加的健壮。


Flink简直太妙了,太牛逼了!