[尚硅谷flink] 检查点笔记

在Flink中,有一套完整的容错机制来保证故障后的恢复,其中最重要的就是检查点。


文章目录

      • 11.1 检查点
        • 11.1.1 检查点的保存
          • 1)周期性的触发保存
          • 2)保存的时间点
          • 3)保存的具体流程
        • 11.1.2 从检查点恢复状态
        • 11.1.3 检查点算法(状态保存的具体算法)
          • 1 检查点分界线(Barrier)
          • 2 Barrier对齐的精准一次
          • 3 Barrier对齐的至少一次
          • 4 非Barrier对齐的精准一次(flink 1.11 之后出现)
        • 11.1.4 检查点配置
          • 1 启用检查点
          • 2 检查点存储
          • 3 其它高级配置
          • 4 demo代码
          • 5 **通用增量checkpoint (changelog)**
          • 6 最终检查点
        • 11.1.5 检查点总结

11.1 检查点

在流处理中,我们可以用存档读档的思路,就是将之前某个时间点所有的状态保存下来,这份“存档”就是所谓的“检查点”(checkpoint)。
image.png
遇到故障重启的时候,我们可以从检查点中“读档”,恢复出之前的状态,这样就可以回到当时保存的一刻接着处理数据了。
:::info
这里所谓的“检查”,其实是针对故障恢复的结果而言的:故障恢复之后继续处理的结果,应该与发生故障前完全一致,我们需要“检查”结果的正确性。所以,有时又会把checkpoint叫做“一致性检查点”。
:::

11.1.1 检查点的保存
1)周期性的触发保存

“随时存档”确实恢复起来方便,可是需要我们不停地做存档操作。如果每处理一条数据就进行检查点的保存,当大量数据同时到来时,就会耗费很多资源来频繁做检查点,数据处理的速度就会受到影响。所以在Flink中,检查点的保存是周期性触发的,间隔时间可以进行设置。

2)保存的时间点

我们应该在所有任务(算子)都恰好处理完一个相同的输入数据的时候,将它们的状态保存下来。
这样做可以实现一个数据被所有任务(算子)完整地处理完,状态得到了保存。
:::info
如果出现故障,我们恢复到之前保存的状态,故障时正在处理的所有数据都需要重新处理;我们只需要让源(source)任务向数据源重新提交偏移量、请求重放数据就可以了。
:::
当然这需要源任务可以把偏移量作为算子状态保存下来,而且外部数据源能够重置偏移量;kafka就是满足这些要求的一个最好的例子。

3)保存的具体流程

检查点的保存,最关键的就是要等所有任务将“同一个数据”处理完毕。下面我们通过一个具体的例子,来详细描述一下检查点具体的保存过程。
回忆一下我们最初实现的统计词频的程序——word count。这里为了方便,我们直接从数据源读入已经分开的一个个单词,例如这里输入的是:
“hello”,“world”,“hello”,“flink”,“hello”,“world”,“hello”,“flink”…
我们所需要的就是每个任务都处理完“hello”之后保存自己的状态。

11.1.2 从检查点恢复状态

:::info

  1. 重启应用
  2. 读取检查点,重置状态
  3. 重置偏移量(向外部数据源重新提交偏移量,比如说Kafka)
  4. 继续处理数据
    :::
    image.png
    image.png
    image.png
    image.png
    image.png
11.1.3 检查点算法(状态保存的具体算法)

在Flink中,采用了基于Chandy-Lamport算法的分布式快照,可以在不暂停整体流处理的前提下,将状态备份保存到检查点。

1 检查点分界线(Barrier)

:::info
借鉴水位线的设计,在数据流中插入一个特殊的数据结构,专门用来表示触发检查点保存的时间点。
barrier肯定是从source算子开始。
:::
收到保存检查点的指令后,Source任务可以在当前数据流中插入这个结构;
之后的所有任务只要遇到它就开始对状态做持久化快照保存。
由于数据流是保持顺序依次处理的,因此遇到这个标识就代表之前的数据都处理完了,可以保存一个检查点;而在它之后的数据,引起的状态改变就不会体现在这个检查点中,而需要保存到下一个检查点。
这种特殊的数据形式,把一条流上的数据按照不同的检查点分隔开,所以就叫做检查点的“分界线”(Checkpoint Barrier)。
image.png

2 Barrier对齐的精准一次

watermark指示的是“之前的数据全部到齐了”,而barrier指示的是“之前所有数据的状态更改保存入当前检查点”:它们都是一个“截止时间”的标志。所以在处理多个分区的传递时,也要以是否还会有数据到来作为一个判断标准。
:::info
具体实现上,Flink使用了Chandy-Lamport算法的一种变体,被称为“异步分界线快照”算法。算法的核心就是两个原则:

  1. 当上游任务向多个并行下游任务发送barrier时,需要广播出去;
  2. 而当多个上游任务向同一个下游任务传递分界线时,需要在下游任务执行“分界线对齐”操作,也就是需要等到所有并行分区的barrier都到齐,才可以开始状态的保存。
    :::
    检查点保存算法具体过程为:
    image.png
    image.png
    image.png
    image.png
    (1)触发检查点:JobManager向Source发送Barrier;
    (2)Barrier发送:向下游广播发送;
    (3)Barrier对齐:下游需要收到上游所有并行度传递过来的Barrier才做自身状态的保存;
    (4)状态保存:有状态的算子将状态保存至持久化。
    (5)先处理缓存数据,然后正常继续处理
    完成检查点保存之后,任务就可以继续正常处理数据了。这时如果有等待分界线对齐时缓存的数据,需要先做处理;然后再按照顺序依次处理新到的数据。当JobManager收到所有任务成功保存状态的信息,就可以确认当前检查点成功保存。之后遇到故障就可以从这里恢复了。
    (补充)由于分界线对齐要求先到达的分区做缓存等待,一定程度上会影响处理的速度;当出现背压时,下游任务会堆积大量的缓冲数据,检查点可能需要很久才可以保存完毕。

为了应对这种场景,Barrier对齐中提供了至少一次语义以及Flink 1.11之后提供了不对齐的检查点保存方式,可以将未处理的缓冲数据也保存进检查点。这样,当我们遇到一个分区barrier时就不需等待对齐,而是可以直接启动状态的保存了。

3 Barrier对齐的至少一次

:::info
精准一次性,在分界线对齐时,barrier 后面的数据需要等待对齐。而至少一次,可以直接进入算子进行计算。
:::

4 非Barrier对齐的精准一次(flink 1.11 之后出现)

:::info
与barrier,在分界线对齐,保存状态的时候,引入了输入缓冲区,以及输出缓冲区,当barrier到来时,不进行计算,直接传入道输出缓冲区的最后一个,然后继续接收数据进行处理,处理完毕后就是输出缓冲区排队。
所以与barrier最大的区别就是在保存状态时,连同缓冲区的数据队列状态一起保存。类似于空间换时间的思想。
:::
image.png
image.png
image.png
image.png

11.1.4 检查点配置

检查点的作用是为了故障恢复,我们不能因为保存检查点占据了大量时间、导致数据处理性能明显降低。为了兼顾容错性和处理性能,我们可以在代码中对检查点进行各种配置。

1 启用检查点

默认情况下,Flink程序是禁用检查点的。如果想要为Flink应用开启自动保存快照的功能,需要在代码中显式地调用执行环境的.enableCheckpointing()方法:

StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
// 每隔1秒启动一次检查点保存
env.enableCheckpointing(1000);

这里需要传入一个长整型的毫秒数,表示周期性保存检查点的间隔时间。如果不传参数直接启用检查点,默认的间隔周期为500毫秒,这种方式已经被弃用。
检查点的间隔时间是对处理性能和故障恢复速度的一个权衡。如果我们希望对性能的影响更小,可以调大间隔时间;而如果希望故障重启后迅速赶上实时的数据处理,就需要将间隔时间设小一些。

2 检查点存储

检查点具体的持久化存储位置,取决于“检查点存储”的设置。默认情况下,检查点存储在JobManager的堆内存中。而对于大状态的持久化保存,Flink也提供了在其他存储位置进行保存的接口。
具体可以通过调用检查点配置的.setCheckpointStorage()来配置,需要传入一个CheckpointStorage的实现类。Flink主要提供了两种CheckpointStorage:作业管理器的堆内存和文件系统。

// 配置存储检查点到JobManager堆内存
env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());

// 配置存储检查点到文件系统
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://namenode:40010/flink/checkpoints"));

对于实际生产应用,我们一般会将CheckpointStorage配置为高可用的分布式文件系统(HDFS,S3等)。

3 其它高级配置

检查点还有很多可以配置的选项,可以通过获取检查点配置(CheckpointConfig)来进行设置。

CheckpointConfig checkpointConfig = env.getCheckpointConfig();

1)常用高级配置

  • 检查点模式(CheckpointingMode)

设置检查点一致性的保证级别,有“精确一次”(exactly-once)和“至少一次”(at-least-once)两个选项。默认级别为exactly-once,而对于大多数低延迟的流处理程序,at-least-once就够用了,而且处理效率会更高。

  • 超时时间(checkpointTimeout)

用于指定检查点保存的超时时间,超时没完成就会被丢弃掉。传入一个长整型毫秒数作为参数,表示超时时间。

  • 最小间隔时间(minPauseBetweenCheckpoints)

用于指定在上一个检查点完成之后,检查点协调器最快等多久可以出发保存下一个检查点的指令。这就意味着即使已经达到了周期触发的时间点,只要距离上一个检查点完成的间隔不够,就依然不能开启下一次检查点的保存。这就为正常处理数据留下了充足的间隙。当指定这个参数时,实际并发为1。

  • 最大并发检查点数量(maxConcurrentCheckpoints)

用于指定运行中的检查点最多可以有多少个。由于每个任务的处理进度不同,完全可能出现后面的任务还没完成前一个检查点的保存、前面任务已经开始保存下一个检查点了。这个参数就是限制同时进行的最大数量。

  • 开启外部持久化存储(enableExternalizedCheckpoints)

用于开启检查点的外部持久化,而且默认在作业失败的时候不会自动清理,如果想释放空间需要自己手工清理。里面传入的参数ExternalizedCheckpointCleanup指定了当作业取消的时候外部的检查点该如何清理。

DELETE_ON_CANCELLATION:在作业取消的时候会自动删除外部检查点,但是如果是作业失败退出,则会保留检查点。

RETAIN_ON_CANCELLATION:作业取消的时候也会保留外部检查点。

  • 检查点连续失败次数(tolerableCheckpointFailureNumber)

用于指定检查点连续失败的次数,当达到这个次数,作业就失败退出。默认为0,这意味着不能容忍检查点失败,并且作业将在第一次报告检查点失败时失败。
2)开启非对齐检查点

  • 非对齐检查点(enableUnalignedCheckpoints)

不再执行检查点的分界线对齐操作,启用之后可以大大减少产生背压时的检查点保存时间。这个设置要求检查点模式(CheckpointingMode)必须为exctly-once,并且最大并发的检查点个数为1。

  • 对齐检查点超时时间(alignedCheckpointTimeout)

该参数只有在启用非对齐检查点的时候有效。参数默认是0,表示一开始就直接用非对齐检查点。如果设置大于0,一开始会使用对齐的检查点,当对齐时间超过该参数设定的时间,则会自动切换成非对齐检查点。

4 demo代码
public class CheckPointConfig {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();

// 指定本地WEB-UI端口号
        configuration.setInteger(RestOptions.PORT, 8082);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.setParallelism(1);

        // 代码中用到hdfs,需要导入hadoop依赖、指定访问hdfs的用户名
        System.setProperty("HADOOP_USER_NAME", "atguigu");

        // TODO 检查点配置
        // 1、启用检查点: 默认是barrier对齐的,周期为5s, 精准一次
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 2、指定检查点的存储位置
//        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/chk");
        checkpointConfig.setCheckpointStorage("file:///Users/xx/project/big_data/flink/flink_study/chk");
        // 3、checkpoint的超时时间: 默认10分钟
        checkpointConfig.setCheckpointTimeout(60000);
        // 4、同时运行中的checkpoint的最大数量
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        // 5、最小等待间隔: 上一轮checkpoint结束 到 下一轮checkpoint开始 之间的间隔,设置了>0,并发就会变成1
        checkpointConfig.setMinPauseBetweenCheckpoints(1000);
        // 6、取消作业时,checkpoint的数据 是否保留在外部系统
        // DELETE_ON_CANCELLATION:主动cancel时,删除存在外部系统的chk-xx目录 (如果是程序突然挂掉,不会删)
        // RETAIN_ON_CANCELLATION:主动cancel时,外部系统的chk-xx目录会保存下来
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        // 7、允许 checkpoint 连续失败的次数,默认0--》表示checkpoint一失败,job就挂掉
        checkpointConfig.setTolerableCheckpointFailureNumber(10);

        // TODO 开启 非对齐检查点(barrier非对齐)
        // 开启的要求: Checkpoint模式必须是精准一次,最大并发必须设为1
        checkpointConfig.enableUnalignedCheckpoints();
        // 开启非对齐检查点才生效: 默认0,表示一开始就直接用 非对齐的检查点
        // 如果大于0, 一开始用 对齐的检查点(barrier对齐), 对齐的时间超过这个参数,自动切换成 非对齐检查点(barrier非对齐)
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(1));


        env
            .socketTextStream("localhost", 7777)
            .flatMap(
                (String value, Collector<Tuple2<String, Integer>> out) -> {
                    String[] words = value.split(" ");
                    for (String word : words) {
                        out.collect(Tuple2.of(word, 1));
                    }
                }
            )
            .returns(Types.TUPLE(Types.STRING, Types.INT))
            .keyBy(value -> value.f0)
            .sum(1)
            .print();

        env.execute();
    }
}
5 通用增量checkpoint (changelog)

了解

6 最终检查点

如果数据源是有界的,就可能出现部分Task已经处理完所有数据,变成finished状态,不继续工作。从Flink 1.14 开始,这些finished状态的Task,也可以继续执行检查点。自1.15 起默认启用此功能,并且可以通过功能标志禁用它:

Configuration config = new Configuration();
config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, false);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
11.1.5 检查点总结
  • Barrier对齐: 一个Task 收到 所有上游 同一个编号的 barrier之后,才会对自己的本地状态做备份
    • 精准一次:在barrier对齐过程中,barrier后面的数据 阻塞等待(不会越过barrier)
    • 至少一次:在barrier对齐过程中,先到的barrier,其后面的数据 不阻塞 接着计算
  • 非Barrier 对齐:一个Task 收到 第一个 barrier时,就开始 执行备份,能保证 精准一次(fhink 1.11出的新算法)

先到的barrier,将 本地状态 备份,其后面的数据接着计算输出未到的barrier,其 前面的数据 接着计算输出,同时 也保存到备份中最后一个barrier到达 该Task时,这个Task的备份结束

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/543694.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

linux 内存寻址

&#xff08;持续更新&#xff09; 相关概念 查看的书籍为 深入linux内核 内存地址 当使用80x86&#xff08;32位&#xff09;微处理器时&#xff0c;一般分为三种不同的地址&#xff1a; 逻辑地址 包含在机器语言指令中用来指定一个操作数或一条指令的地址。每一个逻辑地址…

【服务器配置】Portainer环境配置

Portainer环境配置 概述 Portainer 是一种用于管理 Docker 和 Kubernetes 容器的开源工具。通过其用户友好的 Web 界面&#xff0c;用户可以轻松管理容器、镜像、网络和卷等资源 拉去最新的Portainer docker pull portainer/portainer 安装和启动 docker run -d --restarta…

WindowsServer 2022 AD域控-006-安装副域控

试验拓扑图&#xff1a; 一、测试单域控故障&#xff0c;用户无法修改密码&#xff1b; 域控断网&#xff0c;Win10测试; 二、WindowsServer2022 DC02加入域控&#xff1b; 加入成功 此时域控上只有DC02这台服务器&#xff0c;但DC02并不是域控&#xff1b; 三、WindowsS…

『VUE』17. Dom与模板引用(详细图文注释)

目录 回顾之前的操作ref 属性借助dom使用原生js总结 欢迎关注 『VUE』 专栏&#xff0c;持续更新中 欢迎关注 『VUE』 专栏&#xff0c;持续更新中 回顾之前的操作 之前的这些操作都是我们使用vue为我们渲染的对象,再来操作dom 内容改变{{ 模板语法 }}属性改变 v-bind:添加事…

Java 中文官方教程 2022 版(二十九)

原文&#xff1a;docs.oracle.com/javase/tutorial/reallybigindex.html BCP 47 扩展 原文&#xff1a;docs.oracle.com/javase/tutorial/i18n/locale/extensions.html Java SE 7 版本符合 IETF BCP 47 标准&#xff0c;支持向Locale添加扩展。任何单个字符都可以用于表示扩展&…

2. Spring的创建和Bean的存取

经过前面的学习我们已经大体明白了 IOC 思想以及它的实现方式 DI &#xff0c;本节要讲的是如何Spring框架实现实现DI。 本节目标&#xff1a; Spring(Core) 项目创建将对象存储到 Spring 中将对象(bean)从 Spring 中取出 1. 创建 Spring 项目 与开篇演示的 Spring Boot 项目不…

2024MathorCup数学建模B题成品论文26页+1-4小问代码全解析+答疑

B题 甲骨文智能识别中原始拓片单字自动分割与识别研究 &#xff08;完整版见文末&#xff09; 甲骨文是我国目前已知的最早成熟的文字系统&#xff0c;它是一种刻在龟甲或兽骨上的古老文字。甲骨文具有 极其重要的研究价值&#xff0c;不仅对中国文明的起源具有重要意义&#x…

解放双手,批量绕过403

将dirsearch扫描出来的结果复制到url.txt&#xff0c;如下所示 url.txt [21:18:16] 502 - 0B - /var/log/exception.log [21:18:21] 502 - 0B - /WEB-INF/jetty-env.xml [21:18:22] 502 - 0B - /WEB-INF/weblogic.xml [21:18:27] 502 - 0B - /wp-json/wp/v2/u…

云笔记小程序的实现

1.前言 云笔记, 是基于HotApp小程序统计云后台提供的api接口开发的一个微信小程序。 2.功能 离线保存笔记 云端数据同步, 更换了设备也可以找到以前的笔记 接入了好推二维码提供的数据统计工具, 可以到平台上查看用户分析、留存分析、事件分析。 3.界面效果 ***HotApp云笔…

Java 入门教程||Java 关键字

Java 关键字 Java教程 - Java关键字 Java中的关键字完整列表 关键词是其含义由编程语言定义的词。 Java关键字和保留字&#xff1a; abstract class extends implements null strictfp true assert const false import package super try …

OpenHarmony实战开发-Actor并发模型对比内存共享并发模型

内存共享并发模型指多线程同时执行复数任务&#xff0c;这些线程依赖同一内存并且都有权限访问&#xff0c;线程访问内存前需要抢占并锁定内存的使用权&#xff0c;没有抢占到内存的线程需要等待其他线程释放使用权再执行。 Actor并发模型每一个线程都是一个独立Actor&#xf…

【vs2019】window10环境变量设置

【vs2019】window10环境变量设置 【先赞后看养成习惯】求关注点赞收藏&#x1f60a; 安装VS2019时建议默认安装地址&#xff0c;最好不要改动&#xff0c;不然容易出问题 以下是安装完VS2019后环境变量的设置情况&#xff0c;C:\Program Files (x86)\Microsoft Visual Studi…

20240414,类的嵌套,分文件实现

笑死&#xff0c;和宝哥同时生病了 一&#xff0c;封装-案例 1.0 立方体类 #include<iostream>//分别用全局函数和成员函数判定立方体是否相等 using namespace std;class Cube { public:int m_area;int m_vol;int geth(){return m_h;}int getl() { return m_l; }int…

【群智能算法改进】一种改进的火鹰优化算法 改进的IFHO算法【Matlab代码#77】

文章目录 【获取资源请见文章第5节&#xff1a;资源获取】1. 原始火鹰优化算法1.1 种群初始化1.2 火鹰点火阶段1.3 猎物移动阶段 2. 改进的火鹰优化算法2.1 Tent映射种群初始化2.2 非线性复合自适应惯性权重随机抉择策略 3. 部分代码展示4. 仿真结果展示5. 资源获取 【获取资源…

大模型实战案例:8卡环境微调马斯克开源大模型 Grok-1

节前&#xff0c;我们星球组织了一场算法岗技术&面试讨论会&#xff0c;邀请了一些互联网大厂朋友、参加社招和校招面试的同学&#xff0c;针对算法岗技术趋势、大模型落地项目经验分享、新手如何入门算法岗、该如何准备、面试常考点分享等热门话题进行了深入的讨论。 汇总…

【LeetCode: 705. 设计哈希集合 + 数据结构设计】

&#x1f680; 算法题 &#x1f680; &#x1f332; 算法刷题专栏 | 面试必备算法 | 面试高频算法 &#x1f340; &#x1f332; 越难的东西,越要努力坚持&#xff0c;因为它具有很高的价值&#xff0c;算法就是这样✨ &#x1f332; 作者简介&#xff1a;硕风和炜&#xff0c;…

免费VPS云服务器汇总,最长永久免费使用

目前云服务器市场竞争很激烈&#xff0c;为了方便吸引上云&#xff0c;很多云计算服务商提供免费试用云服务器&#xff0c;下面给大家整理汇总一下免费VPS云服务器&#xff0c;最长永久免费使用&#xff01; 一、雨云&#xff08;优惠码:ABC&#xff09; 活动地址&#xff1a;…

通讯录的实现(顺序表版本)

我们知道通讯录是基于顺序表的前提下&#xff0c;要写好通讯录我们就要深入了解好顺序表。我们先来看看什么是顺序表。&#xff08;注意今天代码量有点多&#xff0c;坚持一下&#xff09;。冲啊&#xff01;兄弟们&#xff01; 顺序表的简单理解 对于顺序表&#xff0c;我们首…

软件测试 测试开发丨Pytest结合数据驱动-yaml,熬夜整理蚂蚁金服软件测试高级笔试题

编程语言 languages: PHPJavaPython book: Python入门: # 书籍名称 price: 55.5 author: Lily available: True repertory: 20 date: 2018-02-17 Java入门: price: 60 author: Lily available: False repertory: Null date: 2018-05-11 yaml 文件使用 查看 yaml 文件 pych…

2024.4.19 Python爬虫复习day07 可视化3

综合案例 需求: 已知2020年疫情数据,都是json数据,需要从文件中读出,进行处理和分析,最终实现数据可视化折线图 相关知识点: json json简介: 本质是一个特定格式的字符串 举例: [{},{},{}] 或者 {}python中json包: import jsonpython数据转为json数据: 变量接收json…