Flink checkpoint 源码分析- Checkpoint snapshot 处理流程

背景

在上一篇博客中我们分析了代码中barrier的是如何流动传递的。Flink checkpoint 源码分析- Checkpoint barrier 传递源码分析-CSDN博客

最后跟踪到了代码org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate#handleEvent 

现在我们接着跟踪相应代码,观察算子接受到了barrier是如何进行下一步代码处理的。以及了解flink应对不同的消费语义(At least once, exactly once)对于checkpoint的影响是怎样的。

代码分析

org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate#handleEvent 中我们主要关注对于checkpointBarrier的处理流程。

processBarrier方法实现上就可以看出,flink barrier的处理分成两种。

在这里我们需要跟踪一下barrierHandler 是如何生成的才能知道后面所要走的流程是哪一步。

通过往上追溯barrierHandler的生成,我们跟踪到方法:org.apache.flink.streaming.runtime.io.checkpointing.InputProcessorUtil#createCheckpointBarrierHandler 从代码中我们可以看到 如果是 EXACTLY_ONCE 那么生成的就SingleCheckpointBarrierHandler, 如果checkpoint 模式是AT_LEAST_ONCE, 生成对应的handler就是CheckpointBarrierTracker。 但是从代码中,EXACTLY_ONCE似乎不是简单的new 一个SingleCheckpointBarrierHandler, 而是通过一个方法来生成。因此需要进一步的观察这个方法是如何实现的。

org.apache.flink.streaming.runtime.io.checkpointing.InputProcessorUtil#createBarrierHandler

这里针对checkpoint类型做了区分,主要是分为aligned checkpoint 和 unaliged checkpoint的差异。这里可以进一步观察一下这两类checkpoint之前的差异。

对比这两个方法参数的差异,发现主要就是两处处参数有差异。subTaskCheckpointCoordinator、barrierHandlerState。这两个的差异主要体现在flink 在aligned checkpoint超时,会切换为unaligned checkpoint。这里可以先按下不表,回到最开始的处理过程。

总结一下就是如果是flink 设置了at least once是使用的是CheckpointBarrierTracker, 当flink模式为exactly once时是SingleCheckpointBarrierHandler。 当为exactly once时checkpoint 类型又可以分为是aligned checkpoint还是unaligned checkpoint。

At least once 下 barrier是如何处理的

at least once 下对于barrier的处理是在以下的方法中实现的。

org.apache.flink.streaming.runtime.io.checkpointing.CheckpointBarrierTracker#processBarrier

public void processBarrier(CheckpointBarrier receivedBarrier, InputChannelInfo channelInfo) throws IOException {final long barrierId = receivedBarrier.getId();// fast path for single channel trackersif (totalNumberOfInputChannels == 1) {markAlignmentStartAndEnd(receivedBarrier.getTimestamp());notifyCheckpoint(receivedBarrier);return;}// general path for multiple input channelsif (LOG.isDebugEnabled()) {LOG.debug("Received barrier for checkpoint {} from channel {}", barrierId, channelInfo);}// find the checkpoint barrier in the queue of pending barriersCheckpointBarrierCount barrierCount = null;int pos = 0;for (CheckpointBarrierCount next : pendingCheckpoints) {if (next.checkpointId == barrierId) {barrierCount = next;break;}pos++;}if (barrierCount != null) {// add one to the count to that barrier and check for completionint numBarriersNew = barrierCount.incrementBarrierCount();if (numBarriersNew == totalNumberOfInputChannels) {// checkpoint can be triggered (or is aborted and all barriers have been seen)// first, remove this checkpoint and all all prior pending// checkpoints (which are now subsumed)for (int i = 0; i <= pos; i++) {pendingCheckpoints.pollFirst();}// notify the listenerif (!barrierCount.isAborted()) {if (LOG.isDebugEnabled()) {LOG.debug("Received all barriers for checkpoint {}", barrierId);}markAlignmentEnd();notifyCheckpoint(receivedBarrier);}}}else {// first barrier for that checkpoint ID// add it only if it is newer than the latest checkpoint.// if it is not newer than the latest checkpoint ID, then there cannot be a// successful checkpoint for that ID anywaysif (barrierId > latestPendingCheckpointID) {markAlignmentStart(receivedBarrier.getTimestamp());latestPendingCheckpointID = barrierId;pendingCheckpoints.addLast(new CheckpointBarrierCount(barrierId));// make sure we do not track too many checkpointsif (pendingCheckpoints.size() > MAX_CHECKPOINTS_TO_TRACK) {pendingCheckpoints.pollFirst();}}}}

如果只有一个inputchannel的情况下,在收到这一个barrier的时候,就可以做snapshot.

在这个中间会经过triggerCheckpointOnBarrier 等方法, 最后实际还是调到了org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl#checkpointState ,看到这里其实这很长的链路实际是一个循环,下一个算子会生成barrier,接着传递这个barrier。

实际情况是作业并行度不唯一,一个subtask往往是有多个inputchannel. 可以继续看看是如何处理的。

这里面当收取到第一个barrier,会将这个barrier信息存在一个队列中。

每当收到一个barrier的时候会进行计数,当收取到的是最后一个barrier的时候把之前的barrier全部清除,之后就可以通知做checkpoint snapshot, 这个流程就和之前的一个信道的checkpoint流程是一致的。

总结而言:at least 类型的checkpoint是在收到最后一个barrier的时候开始做snapshot的。

Exactly once checkpoint是如何处理的

首先看这一段的代码

@Overridepublic void processBarrier(CheckpointBarrier barrier, InputChannelInfo channelInfo) throws IOException {long barrierId = barrier.getId();LOG.debug("{}: Received barrier from channel {} @ {}.", taskName, channelInfo, barrierId);if (currentCheckpointId > barrierId|| (currentCheckpointId == barrierId && !isCheckpointPending())) {if (!barrier.getCheckpointOptions().isUnalignedCheckpoint()) {inputs[channelInfo.getGateIdx()].resumeConsumption(channelInfo);}return;}checkNewCheckpoint(barrier);checkState(currentCheckpointId == barrierId);if (numBarriersReceived++ == 0) {if (getNumOpenChannels() == 1) {markAlignmentStartAndEnd(barrier.getTimestamp());} else {markAlignmentStart(barrier.getTimestamp());}}// we must mark alignment end before calling currentState.barrierReceived which might// trigger a checkpoint with unfinished future for alignment durationif (numBarriersReceived == numOpenChannels) {if (getNumOpenChannels() > 1) {markAlignmentEnd();}}try {currentState = currentState.barrierReceived(context, channelInfo, barrier);} catch (CheckpointException e) {abortInternal(barrier.getId(), e);} catch (Exception e) {ExceptionUtils.rethrowIOException(e);}if (numBarriersReceived == numOpenChannels) {numBarriersReceived = 0;lastCancelledOrCompletedCheckpointId = currentCheckpointId;LOG.debug("{}: Received all barriers for checkpoint {}.", taskName, currentCheckpointId);resetAlignmentTimer();allBarriersReceivedFuture.complete(null);}}

这里需要关注一下currentState, 在最开始我们看了他的构造函数AlternatingWaitingForFirstBarrier, 因此可以可以看这个方法具体是现实。

这里可以看到这里会block 住收到barrier的信道,如果barrier 都收齐了,之后会检查是不是unaligned的checkpoint, 如果不是可以直接做一次checkpoint。这个checkpoint和之前的流程是一致的。

这里的下一个分支是超时转化,比如设置为30s,前30s是做aligned checkpoint, 如果30s还没有完成,就会转化为unaligned checkpoint。 当然,你如果不想有超时时间,可以直接设置为0.

如果是unaligned checkpoint, 会将channel 里面的数据也写会到远端。

这个中间会有一些状态转化,每次barrier的到达都会触发不同的状态变化。其中我们看到对于uc来说,uc的第一个barrier到达了,就会触发一次global checkpoint,所以这个时候是不会block住信道的。org.apache.flink.streaming.runtime.io.checkpointing.AlternatingWaitingForFirstBarrierUnaligned#barrierReceived

org.apache.flink.streaming.runtime.io.checkpointing.AlternatingCollectingBarriersUnaligned#barrierReceived

最后如果收到所有的barrier之后会finished checkpoint。状态恢复到原位。

总结一下:在exactly once的语义下,aligned checkpoint的做法是,收到一个barrier的时候会将对应的channel block住。当收到最后一个barrier的时候再做一次checkpoint。

unaligned的做法是,收到barrier的时候,第一步就会触发一次checkpoint, 之后会不断上传channel state, 当收到最后一个barrier则表示checkpoint结束。

在这一篇文章我们主要介绍了ssubtask级别的snapshot的,后面再进一步介绍一下整体流程,就可以结束相关介绍了。

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

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

面向电商家居行业3D室内场景合成中的空间感知

本文主要介绍了3D场景合成技术在电商领域&#xff0c;尤其是家居家装行业的应用。它解释了如何使用3D场景合成创建逼真的室内设计&#xff0c;让消费者能够交互式地查看和体验产品&#xff0c;提高购物的趣味性和效率。文章提到了两种主要的3D室内场景生成算法&#xff1a;传统…

Electron | 桌面应用的开发神器

初探 Electron 教程将介绍 Electron 打包应用的全过程&#xff0c;从本地测试&#xff0c;打包&#xff0c;到 GitHub 自动化。讲解 Electron Forge 和 Electron Builder 的用法&#xff0c;以及如何在 GitHub Actions 中自动化生成和发布应用。 官方资源 Electron Document…

Vue.js【路由】

初识路由 提到路由&#xff08;Route&#xff09;&#xff0c;一般我们会联想到网络中常见的路由器&#xff08;Router&#xff09;&#xff0c;那么路由和路由器之间有什么关联呢&#xff1f;路由是指路由器从一个接口接收到数据&#xff0c;根据数据的目的地址将数据定向传送…

Linux开发--Linux内核开发移植

Linux内核开发移植 Linux内核版本变迁及其获得 Linux是最受欢迎的自由电脑操作系统内核&#xff0c; 是一个用C语言写成&#xff0c; 并且符合POSIX标准的类Unix操作系统 Linux是由芬兰黑客Linus Torvalds开发的&#xff0c; 目的是尝试在英特尔x86架构上提供自由免费的类Un…

IDEA及Maven配置代理及Maven中央仓库配置详解

一、配置代理 首先&#xff0c;需要本地开启代理入口&#xff0c;如图。 这个跟你使用代理软件有关。像我使用的是qv2ray。 其次&#xff0c;idea配置代理&#xff0c;如图。 1.1 idea配置代理 打开Settings&#xff0c;如图 1.2 maven配置代理 maven配置代理&#xff0c;修…

[单机]完美国际_V155_GM工具_VM虚拟机

[端游] 完美国际单机版V155一键端PC电脑网络游戏完美世界幻海凌云家园 本教程仅限学习使用&#xff0c;禁止商用&#xff0c;一切后果与本人无关&#xff0c;此声明具有法律效应&#xff01;&#xff01;&#xff01;&#xff01; 教程是本人亲自搭建成功的&#xff0c;绝对是…

易图讯三维电子沙盘-大数据处理服务

易图讯科技10名高级大数据工程师&#xff0c;高效、快速进行POI、DEM、高清卫星影像、地形地貌、路网、矢量地图等海量大数据处理服务。 免费专业提供POI、AOI、DEM、高清卫星影像、地形地貌、路网、矢量地图等海量大数据处理服务。 1年更新2次POI、高清卫星影像。

Unity基础

概述 基础知识 3D教学 数学计算公共类Mathf 练习: 三角函数 练习&#xff1a; Unity中的坐标系 Vector3向量 向量模长和单位向量 向量加减乘除 练习&#xff1a; 向量点乘 向量叉乘 向量插值运算 Quaternion四元数 为何要使用四元数 四元数是什么 四元数常用方法 四元数计算 练…

怎么制作流程图?介绍制作方法

怎么制作流程图&#xff1f;在日常生活和工作中&#xff0c;流程图已经成为我们不可或缺的工具。无论是项目规划、流程优化&#xff0c;还是学习理解复杂系统&#xff0c;流程图都能帮助我们更直观地理解和表达信息。然而&#xff0c;很多人可能并不清楚&#xff0c;其实制作流…

cmu15445 2023fall project3 详细过程(下)QUERY EXECUTION

QUERY EXECUTION task3/task4 Task #3 - HashJoin Executor and Optimization1、HashJoin1.1 思路1.2 代码 2 NestedLoopJoin优化为HashJoin2.1 思路2.2 代码 Task #4 Sort Limit Executors Top-N Optimization Window Functions1、Sort1.1 思路1.2 代码 2、Limit Executors2…

CCPD车牌检测识别数据集

CCPD 是一个在开源免费的中国城市车牌识别数据集。 1. 介绍 CCPD (Chinese City Parking Dataset, ECCV)是中国城市车牌数据集&#xff0c;共有两个CCPD2019和CCPD2020 数据集&#xff0c;总数据量约35W左右&#xff0c;可用于车牌检测和识别模型算法开发。 CCPD 发表的论文:…

从0开始学python(七)

目录 前言 1 break、continue和pass函数 1.1 break 1.2 continue 1.3 pass 2、序列的索引及切片操作 2.1字符串的索引和切片 2.1.1 字符串索引 2.1.2 字符串切片 总结 前言 上一篇文章我们介绍了python中的循环结构&#xff0c;包括for和while的使用。本章接着往下讲。…

excel常见图表大全

Excel图表是一种以图形形式呈现数据的工具&#xff0c;它将数字和统计信息转化为直观的视觉元素&#xff0c;如线图、柱状图、饼图等。这些图表可以帮助人们更容易地理解数据的趋势、关系和模式。 使用场景 Excel图表广泛应用于各个领域&#xff0c;包括&#xff1a; 商务分…

【NVIDIA Jetpack6.0】Jetson AGX Orin内核、设备树更新指南

博主未授权任何人或组织机构转载博主任何原创文章&#xff0c;感谢各位对原创的支持&#xff01; 博主链接 本人就职于国际知名终端厂商&#xff0c;负责modem芯片研发。 在5G早期负责终端数据业务层、核心网相关的开发工作&#xff0c;目前牵头6G算力网络技术标准研究。 博客…

anconda创建虚拟环境,使用虚拟环境(基于win平台)

假设已经安装了anconda&#xff0c;打开anaconda的 shell。 查看已存在的虚拟环境&#xff0c;base是默认的&#xff0c;不用理会&#xff0c;后面的yolov5就是用户创建的 #查看有那些虚拟环境 (base) PS C:\Users\x> conda info -e # conda environments: # base …

【江科大STM32学习笔记】新建工程

1.建立工程文件夹&#xff0c;Keil中新建工程&#xff0c;选择型号 2.工程文件夹里建立Start、Library、User等文件夹&#xff0c;复制固件库里面的文件到工程文件夹 为添加工程文件准备&#xff0c;建文件夹是因为文件比较多需要分类管理&#xff0c;需要用到的文件一定要复…

wordpress主题 7B2 PRO主题5.4.2免授权直接安装

内容目录 一、详细介绍二、效果展示1.部分代码2.效果图展示 三、学习资料下载 一、详细介绍 WordPress 资讯、资源、社交、商城、圈子、导航等多功能商用主题&#xff1a;B2 PRO 其设计风格专业且时尚&#xff0c;功能十分强大&#xff0c;包括多栏布局、自定义页面、强大的主…

faad2交叉编译——aac解码为pcm,解决faad单通道转双通道问题

FAAD是比较成熟高效的开源AAC解码库&#xff0c;这里用于解码AAC生成PCM数据&#xff0c;用于音频播放。这里因为faad库&#xff0c;会将单通道转化为双通道踩了些坑&#xff0c;所以记录一下。 我使用的是2.11.0版本&#xff0c;貌似往前的版本没有使用CMake&#xff0c;需要c…

网络应用层

叠甲&#xff1a;以下文章主要是依靠我的实际编码学习中总结出来的经验之谈&#xff0c;求逻辑自洽&#xff0c;不能百分百保证正确&#xff0c;有错误、未定义、不合适的内容请尽情指出&#xff01; 文章目录 1.使用协议和序列化1.1.自定义协议&#xff0c;自定义序列化1.2.自…

综合性练习(验证码案例)

目录 一、需求 二、准备工作 三、约定前后端交互接口 1、需求分析 2、接口定义 四、Hutool工具介绍 1、引入依赖 2、测试使用Hutool生成验证码 五、实现服务器端代码 代码解读&#xff1a; 六、调整前端页面代码 七、运行测试 随着安全性的要求越来越高&#xff0c…