Flink-容错机制-检查点相关内容

左手的ㄟ右手 2022-09-09 12:48 236阅读 0赞

## 检查点(checkpoint) ##

### 一致性检查点(Checkpoints) ###

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_17_color_FFFFFF_t_70_g_se_x_16]

**Flink 故障恢复机制的核心,就是应用状态的一致性检查点**

有状态流应用的一致检查点,其实就是所有任务的状态,在某个时间点的一份拷贝(一份快照);这个时间点,应该是所有任务都恰好处理完一个相同的输入数据的时候

以上图为从1开始的整数 数据，sum\_even为偶数求和，sum\_odd为奇数求和，以上已经完成了1-5的相关求和。

### 从检查点恢复状态 ###

![在这里插入图片描述][f72f897b1b434b95bfc5e944e8f6e0ad.png]

在执行流应用程序期间,Flink 会定期保存状态的一致检查点

如果发生故障, Flink 将会使用最近的检查点来一致恢复应用程序的状态,并重新启动处理流程

![在这里插入图片描述][22efd6d5ab364b679c748fc4bd52bc7c.png]

遇到故障之后,第一步就是重启应用

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_19_color_FFFFFF_t_70_g_se_x_16]

第二步是从 checkpoint 中读取状态,将状态重置

从检查点重新启动应用程序后,其内部状态与检查点完成时的状态完全相同

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_19_color_FFFFFF_t_70_g_se_x_16 1]

第三步:开始消费并处理检查点到发生故障之间的所有数据

这种检查点的保存和恢复机制可以为应用程序状态提供“精确一次”(exactly-once，处理一次，且只处理一次，不丢不重复)的一致性,因为所有算子都会保存检查点并恢复其所有状态,这样一来所有的输入流就都会被重置到检查点完成时的位置

### Flink 检查点算法 ###

检查点的实现算法

一种简单的想法

—— 暂停应用,保存状态到检查点,再重新恢复应用

Flink 的改进实现

—— 基于 Chandy-Lamport 算法的分布式快照

—— 将检查点的保存和数据处理分离开,不暂停整个应用

**检查点分界线(Checkpoint Barrier)**

Flink 的检查点算法用到了一种称为**分界线(barrier)的特殊数据形式**,用来把一条流上数据按照不同的检查点分开

分界线之前到来的数据导致的状态更改,都会被包含在当前分界线所属的检查点中;而基于分界线之后的数据导致的所有更改,就会被包含在之后的检查点中

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]

现在是一个有两个输入流的应用程序,用并行的两个 Source 任务来读取

以上图蓝色数据为1-5，黄色数据为1-6，其中蓝1和黄1已经处理完成了得到了蓝2（1+1），并且黄3已经处理完成得到了黄5（2+3）

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]

JobManager 会向每个 source 任务发送一条带有新检查点 ID 的消息,通过这种方式来启动检查点；其中的蓝色三角形是2号检查点；发送的信息是同时发送给所有的Source的；在蓝3蓝4之间以及黄4黄5之间会插入checkpoint；

此图比较于上图的区别在与蓝2和黄2已经备sink输出了。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 2]

数据源将它们的状态写入检查点（蓝3，黄4）,并发出一个检查点 barrier，发送barrier时是广播发送的。

状态后端在状态存入检查点之后,会返回通知给 source 任务,source 任务就会向 JobManager 确认检查点完成；

以上图中蓝3已经处理完毕了；黄4数据先被发送，然后广播黄4对应的barrier。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 3]

分界线对齐:barrier 向下游传递,sum 任务会等待所有输入分区的 barrier 到达

对于barrier已经到达的分区,继续到达的数据会被缓存，蓝4将会被暂时缓存，不做计算

而barrier尚未到达的分区,数据会被正常处理

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 4]

当收到所有输入分区的 barrier 时,任务就将其状态保存到状态后端的检查点中,然后将 barrier 继续向下游转发

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 5]

向下游转发检查点 barrier 后,任务继续正常的数据处理

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 6]

Sink 任务向 JobManager 确认状态保存到 checkpoint 完毕

当所有任务都确认已成功将状态保存到检查点时,检查点就真正完成了

### 保存点(Savepoints) ###

Flink 还提供了可以自定义的镜像保存功能,就是**保存点(savepoints)**

原则上,创建保存点使用的算法与检查点完全相同,因此保存点可以认为就是具有一些额外元数据的检查点

Flink不会自动创建保存点,因此用户(或者外部调度程序)必须明确地触发创建操作

保存点是一个强大的功能。除了故障恢复外,保存点可以用于:有计划的手动备份,更新应用程序,版本迁移,暂停和重启应用,等等

检查点配置相关示例

// 状态后端配置
    // env.setStateBackend(new MemoryStateBackend());
    env.setStateBackend(new FsStateBackend(""));
    // env.setStateBackend(new RocksDBStateBackend(""));
    
    // 检查点配置
    env.enableCheckpointing(1000); // 默认500ms
    
    // 高级选项
    // 检查点模式，默认就是精确一次
    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    // 检查点超时时间，默认10分钟
    env.getCheckpointConfig().setCheckpointTimeout(60000);
    // 最大同时进行的checkpoint，就是前一个checkpoint还没有处理完，后一个就开始了，默认1
    env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
    // 前一次checkpoint保存成功之后，下一次checkpoint执行的最小间隔时间，默认0
    env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100);
    // 其他不常用配置
    // 当恢复数据的时候是否只选择checkpoint来进行恢复，默认是false，将会选择checkpoint和savepoint中更近的来恢复
    env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
    // 允许checkpoint失败多少次，默认是0次
    env.getCheckpointConfig().setTolerableCheckpointFailureNumber(1);
    
    // 重启策略的配置
    // 固定延时重启，一个是重启次数，第二个是两次重启的间隔时间
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000));
    // 失败率重启，10分钟之内尝试重启3次，每次重启间隔1分钟
    env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(10), Time.minutes(1)));

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_17_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/2e2f2273ee9f483ca81f2568ec1debf9.png
[f72f897b1b434b95bfc5e944e8f6e0ad.png]: /images/20220829/5cc519a5f437421ea9d7983901aa891c.png
[22efd6d5ab364b679c748fc4bd52bc7c.png]: /images/20220829/88f9d34fc3224ce1b051bf1c0aaf9b6c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_19_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/c07328ccce7b4cc5ba63f811929f27e0.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_19_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/6cedb28c74ec4c05a85efa4e575f2978.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/888593f9d43a4dcebd5cc2598b78ecd0.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/7ee7a68ccafd40e3972cd3aee3121bb7.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/f2eaba5256c3445bbe181c3ab1428aa9.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/1019e3bc79914174b4beb73d8cccc686.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220829/1a332e402a5748b9b0e178c8dac9a04f.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220829/035f361464064c12b7c0e24720b006b4.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 6]: /images/20220829/02e27be388014de88adf3d6ce41d18a8.png