当前位置: 首页 > news >正文

Spark 3.1.1 shuffle fetch 导致shuffle错位的问题

背景

最近从数据仓库小组那边反馈了一个问题,一个SQL任务出来的结果不正确,重新运行一次之后就没问题了,具体的SQL如下:

select col1,count(1) as cnt
from table1
where dt = '20230202' 
group by col1
having count(1) > 1

这个问题是偶发的,在其运行的日志中会发现如下三类日志:

FetchFailed 
TaskKilled (another attempt succeeded)
ERROR (org.apache.spark.network.shuffle.RetryingBlockFetcher:231) - Failed to fetch block shuffle_4865_2481
283_286, and will not retry (3 retries)

最终在各种同事的努力下,找到了一个Jira:SPARK-34534

分析

直接切入主题,找到对应的类OneForOneBlockFetcher,该类会被NettyBlockTransferService(没开启ESS)和ExternalBlockStoreClient(开启ESS)调用,其中start方法:

public void start() {client.sendRpc(message.toByteBuffer(), new RpcResponseCallback() {@Overridepublic void onSuccess(ByteBuffer response) {try {streamHandle = (StreamHandle) BlockTransferMessage.Decoder.fromByteBuffer(response);logger.trace("Successfully opened blocks {}, preparing to fetch chunks.", streamHandle);// Immediately request all chunks -- we expect that the total size of the request is// reasonable due to higher level chunking in [[ShuffleBlockFetcherIterator]].for (int i = 0; i < streamHandle.numChunks; i++) {if (downloadFileManager != null) {client.stream(OneForOneStreamManager.genStreamChunkId(streamHandle.streamId, i),new DownloadCallback(i));} else {client.fetchChunk(streamHandle.streamId, i, chunkCallback);}}} catch (Exception e) {logger.error("Failed while starting block fetches after success", e);failRemainingBlocks(blockIds, e);}}@Overridepublic void onFailure(Throwable e) {logger.error("Failed while starting block fetches", e);failRemainingBlocks(blockIds, e);}});}

其中的message的初始化在构造方法中:

 if (!transportConf.useOldFetchProtocol() && isShuffleBlocks(blockIds)) {this.message = createFetchShuffleBlocksMsg(appId, execId, blockIds);} else {this.message = new OpenBlocks(appId, execId, blockIds);}

其中transportConf.useOldFetchProtocol 也就是 spark.shuffle.useOldFetchProtocol配置(默认是false),如果是shuffle block的话,就会运行到:createFetchShuffleBlocksMsg方法,对于为什么存在这么一个判断,具体参考SPARK-27665
关键的就是 createFetchShuffleBlocksMsg 方法:
这个方法的作用就是: 构建一个FetchShuffleBlocks(appId, execId, shuffleId, mapIds, reduceIdArr, batchFetchEnabled) 对象,其中里面的值
如图:
在这里插入图片描述
其中这里有一点需要注意:

 long[] mapIds = Longs.toArray(mapIdToReduceIds.keySet());reduceIdArr[i] = Ints.toArray(mapIdToReduceIds.get(mapIds[i]));

这里面对MapIdReduceId 进行了重组(在获得streamHandle的时候内部会根据reduceIdArr构建blocks索引,下文中会说到)会导致和成员变量blockIds的顺序不一致,为什么两者不一致会导致问题呢?
原因在于任务的fetch失败会导致重新进行fetch,如下:

  client.fetchChunk(streamHandle.streamId, i, chunkCallback);

chunkCallback的代码如下:

private class ChunkCallback implements ChunkReceivedCallback {@Overridepublic void onSuccess(int chunkIndex, ManagedBuffer buffer) {// On receipt of a chunk, pass it upwards as a block.listener.onBlockFetchSuccess(blockIds[chunkIndex], buffer);}@Overridepublic void onFailure(int chunkIndex, Throwable e) {// On receipt of a failure, fail every block from chunkIndex onwards.String[] remainingBlockIds = Arrays.copyOfRange(blockIds, chunkIndex, blockIds.length);failRemainingBlocks(remainingBlockIds, e);}}

String[] remainingBlockIds = Arrays.copyOfRange(blockIds, chunkIndex, blockIds.length),此处的chunckIndex就是shuffle blocks的索引下标,也就是下文中numBlockIds组成的数组下标,
但是这个和createFetchShuffleBlocksMsg输出的顺序是不一致的,所以如果发生问题重新fetch的时候,数据有错位,具体可以看:
ShuffleBlockFetcherIterator中的

    if (req.size > maxReqSizeShuffleToMem) {shuffleClient.fetchBlocks(address.host, address.port, address.executorId, blockIds.toArray,blockFetchingListener, this)} else {shuffleClient.fetchBlocks(address.host, address.port, address.executorId, blockIds.toArray,blockFetchingListener, null)}

其中blockFetchingListener回调方法onBlockFetchSuccess会把fetch的block数据和shuffleBlockId一一对应上

ESS端构建blocks的信息

在start方法中,client.sendRpc向对应的ESS发送对应的请求shuffle数据信息,ESS会重新构建blocks的信息,组成StreamHandle(streamId, numBlockIds)返回给请求端:
具体为ExternalBlockHandler的handleMessage方法:

if (msgObj instanceof FetchShuffleBlocks) {FetchShuffleBlocks msg = (FetchShuffleBlocks) msgObj;checkAuth(client, msg.appId);numBlockIds = 0;if (msg.batchFetchEnabled) {numBlockIds = msg.mapIds.length;} else {for (int[] ids: msg.reduceIds) {numBlockIds += ids.length;}}streamId = streamManager.registerStream(client.getClientId(),new ShuffleManagedBufferIterator(msg), client.getChannel());
。。。
callback.onSuccess(new StreamHandle(streamId, numBlockIds).toByteBuffer());

这里的numBlockIds就是OneForOneBlockFetcher中的streamHandle.numChunks
如图:在这里插入图片描述

没有开启ESS端的构建blocks的信息

这里和上面的一样,只不过对应的方法为NettyBlockRpcServerreceive:

      case fetchShuffleBlocks: FetchShuffleBlocks =>val blocks = fetchShuffleBlocks.mapIds.zipWithIndex.flatMap { case (mapId, index) =>if (!fetchShuffleBlocks.batchFetchEnabled) {fetchShuffleBlocks.reduceIds(index).map { reduceId =>blockManager.getLocalBlockData(ShuffleBlockId(fetchShuffleBlocks.shuffleId, mapId, reduceId))}} else {val startAndEndId = fetchShuffleBlocks.reduceIds(index)if (startAndEndId.length != 2) {throw new IllegalStateException(s"Invalid shuffle fetch request when batch mode " +s"is enabled: $fetchShuffleBlocks")}Array(blockManager.getLocalBlockData(ShuffleBlockBatchId(fetchShuffleBlocks.shuffleId, mapId, startAndEndId(0), startAndEndId(1))))}}val numBlockIds = if (fetchShuffleBlocks.batchFetchEnabled) {fetchShuffleBlocks.mapIds.length} else {fetchShuffleBlocks.reduceIds.map(_.length).sum}val streamId = streamManager.registerStream(appId, blocks.iterator.asJava,client.getChannel)logTrace(s"Registered streamId $streamId with $numBlockIds buffers")responseContext.onSuccess(new StreamHandle(streamId, numBlockIds).toByteBuffer)

这里的numBlockIds就是OneForOneBlockFetcher中的streamHandle.numChunks
如图:
在这里插入图片描述
所以在以上两种情况下,只要有重新fetch数据的操作,就会存在数据的错位,导致数据的不准确

解决

直接git cherry-pick对应的commit就行:

git cherry-pick 4e438196114eff2e1fc4dd726fdc1bda1af267da

相关文章:

Spark 3.1.1 shuffle fetch 导致shuffle错位的问题

背景 最近从数据仓库小组那边反馈了一个问题,一个SQL任务出来的结果不正确&#xff0c;重新运行一次之后就没问题了&#xff0c;具体的SQL如下&#xff1a; select col1,count(1) as cnt from table1 where dt 20230202 group by col1 having count(1) > 1这个问题是偶发…...

2月第2周榜单丨飞瓜数据B站UP主排行榜(哔哩哔哩平台)发布!

飞瓜轻数发布2023年2月6日-2月12日飞瓜数据UP主排行榜&#xff08;B站平台&#xff09;&#xff0c;通过充电数、涨粉数、成长指数三个维度来体现UP主账号成长的情况&#xff0c;为用户提供B站号综合价值的数据参考&#xff0c;根据UP主成长情况用户能够快速找到运营能力强的B站…...

Jdk19 动态编译 Java源码为 Class 文件

动态编译 Java 源码为 Class一.背景1.Jdk 版本2.需求二.Java 源码动态编译实现1.Maven 依赖2.源码包装类3.Java 文件对象封装类4.文件管理器封装类5.类加载器6.类编译器三.动态编译测试1.普通测试类2.接口实现类3.测试四.用动态编译 Class 替换 SpringBoot 的 Bean&#xff08;…...

安装 GPU 版本的 tensorflow 完整版本

前言&#xff1a; 之前安装的 CPU 版本的 tensorflow 一直出问题&#xff0c;索性就直接安装 GPU 版本的 tensorflow 了&#xff08;有了GPU 就不能浪费&#xff09;。 安装过程&#xff1a; 1&#xff09;看自己有无 GPU&#xff0c;找到对应 GPU 的版本&#xff1a;任务管理…...

BOM编程-设置地址栏上的URL

<!DOCTYPE html> <html> <head> <meta charset"utf-8"> <title>设置地址栏上的URL</title> </head> <body> <script> function go(){ // 获…...

设计模式之原型模式与建造者模式详解和应用

目录1 原型模式1.1 原型模式定义1.2 原型模式的应用场景1.3 原型模式的通用写法&#xff08;浅拷贝&#xff09;1.4 使用序列化实现深度克隆1.5 克隆破坏单例模式1.6 原型模式在源码中的应用1.7 原型模式的优缺点1.8 总结2 建造者模式2.1 建造者模式定义2.2 建造者模式的应用场…...

C语言(函数和递归)

函数是完成特定任务的独立程序代码单元。 目录 一.函数 1.创建一个简单的函数 2.定义带形式参数的函数 3.使用return从函数中返回值 二.递归 一.函数 1.创建一个简单的函数 #include <stdio.h> void print(void); //函数原型 int main(){ print(); //函…...

快乐的shell命令行

快乐的shell命令行 PART1——基础 1.权限 #超级用户权限$普通用户 2.复制粘贴 复制&#xff1a;鼠标左键沿着文本拖动高亮的文本被复制到X管理的缓冲区&#xff08;或者双击一个单词&#xff09;粘贴&#xff1a;鼠标中键 3.简单命令 时间和日期date当前月份的日历cal磁…...

大数据面试题flume篇

1.Flume 的Source&#xff0c;Sink&#xff0c;Channel 的作用&#xff1f;你们Source 是什么类型&#xff1f; 1. 作用 &#xff08;1&#xff09;Source组件是专门用来收集数据的&#xff0c;可以处理各种类型、各种格式的日志数据&#xff0c;包括 avro、thrift、exec、jm…...

零信任-深信服零信任aTrust介绍(5)

​深信服零信任aTrust介绍 深信服是国内领先的互联网信任服务提供商&#xff0c;也是国内首家通过认证的全球信任服务商。深信服零信任是其中一项核心的信任技术&#xff0c;主要针对身份认证、数字签名、数字证书等方面的信任问题。 深信服零信任提供了一种新的安全保护模式…...

UVa 1343 The Rotation Game 旋转游戏 IDA* BFS 路径还原

题目链接&#xff1a;The Rotation Game 题目描述&#xff1a; 给定二十四个整数&#xff0c;这二十四个整数由八个一&#xff0c;八个二&#xff0c;八个三组成&#xff0c;从左到右&#xff0c;从上到下依次描述下图方格中的数字&#xff1a; 例如上图左边对应的输入就是[1,…...

硬件学习 软件Cadence day02 画原理图的基本操作 (键盘快捷键 , 原理图设计流程 , 从开始到导出网表流程)

1. ORCAD Capture cls 界面的快捷键 键盘 按键对应的操作I放大 &#xff08;可以滚轮操作&#xff09;O缩小 &#xff08;可以滚轮操作&#xff09;W画线Esc退出现在的状态 &#xff08;画图界面 右键 End xxx&#xff09;N放置网络标号J放置节点 (控制…...

【python】基于Socket的聊天室Python开发

基于Socket的聊天室Python开发一、Socket简述二、创建服务端Server2.1 创建服务端初始化2.2 监听客户端连接2.3 处理客户端消息三、创建客户端Client3.1 创建服务端初始化3.2 发送消息3.3 接收消息3.3 线程工作3.4 线程工作是不是挺好玩的呢&#xff1f;也可以作为课程设计哦&a…...

2023想转行软件测试的看过来,你想要了解的薪资、前景、岗位方向、学习路线都讲明白了

在过去的一年中&#xff0c;软件测试行业发展迅速&#xff0c;随着数字化技术应用的广泛普及&#xff0c;业界对于软件测试的要求也在持续迭代与增加。 同样的&#xff0c;有市场就有需求&#xff0c;软件测试逐渐成为企业中不可或缺的岗位&#xff0c;作为一个高薪又需求广的…...

TortoiseSVN的使用

基本概念 版本库 SVN保持数据的地方&#xff0c;所有的文件都保存在这个库中&#xff0c;Tortoise访问的就是远程服务器上的Subversion版本库。 工作拷贝 就是工作副本&#xff0c;可将版本库的文件拷贝到本地中&#xff0c;可以任意修改&#xff0c; 不会影响版本库。在你…...

操作系统(day09) -- 连续分配管理方式

连续分配管理方式 单元连续分配 动态分区分配 1.系统要用什么样的数据结构记录内存的使用情况&#xff1f; 两种常用的数据结构 空闲分区表 每个空闲分区对应一个表项。表项中包含分区号、分区大小、分区起始地址等信息空闲分区链 每个分区的起始部分和末尾部分分别设置前向…...

APISpace 带你一起走进西湖美景

俗话说&#xff1a;“上有天堂&#xff0c;下有苏杭”。 “欲把西湖比西子&#xff0c;浓妆艳抹总相宜” 今天我就带大家走进杭州的西湖美景。自古以来&#xff0c;文人歌者面对西湖美景留下千古绝句&#xff0c;还以西湖为背景书写了一段段动人的爱情传说。 天生自带浪漫色…...

傻白探索Chiplet,Design Space Exploration for Chiplet-Assembly-Based Processors(十三)

阅读了Design Space Exploration for Chiplet-Assembly-Based Processors这篇论文&#xff0c;是关于chiplet设计空间探索的&#xff0c;个人感觉核心贡献有两个&#xff1a;1.提出使用整数线性规划算法进行Chiplet的选择&#xff1b;2.基于RE和NRE提出了一个cost模型&#xff…...

系统分析师真题2020试卷相关概念一

对象系统测试的基本概念: 面向对象系统的单元测试包括方法层次的测试、类层次的测试和类树层次的测试。方法层次的测试类似于传统软件测试中对单个函数的测试; 测试技术: 方法层次的测试,单个函数的测试;常用的技术:等价类划分测试、组合功能测试、递归函数的测试和多态…...

20230215_数据库过程_渠道业务计算过程

—20221209 渠道产能 —自有人员工号表 shzc.xc_qdcn_pgtx_opertype —select * from shzc.xc_qdcn_pgtx_opertype for update ; —渠道基础目录 shzc.xc_qdcn_pgtx_qdtype —select * from shzc.xc_qdcn_pgtx_qdtype for update ; SQL_STRING:‘update shzc.xc_qdcn_pgtx_q…...

Java-41 深入浅出 Spring - 声明式事务的支持 事务配置 XML模式 XML+注解模式

点一下关注吧&#xff01;&#xff01;&#xff01;非常感谢&#xff01;&#xff01;持续更新&#xff01;&#xff01;&#xff01; &#x1f680; AI篇持续更新中&#xff01;&#xff08;长期更新&#xff09; 目前2025年06月05日更新到&#xff1a; AI炼丹日志-28 - Aud…...

现代密码学 | 椭圆曲线密码学—附py代码

Elliptic Curve Cryptography 椭圆曲线密码学&#xff08;ECC&#xff09;是一种基于有限域上椭圆曲线数学特性的公钥加密技术。其核心原理涉及椭圆曲线的代数性质、离散对数问题以及有限域上的运算。 椭圆曲线密码学是多种数字签名算法的基础&#xff0c;例如椭圆曲线数字签…...

Axios请求超时重发机制

Axios 超时重新请求实现方案 在 Axios 中实现超时重新请求可以通过以下几种方式&#xff1a; 1. 使用拦截器实现自动重试 import axios from axios;// 创建axios实例 const instance axios.create();// 设置超时时间 instance.defaults.timeout 5000;// 最大重试次数 cons…...

初学 pytest 记录

安装 pip install pytest用例可以是函数也可以是类中的方法 def test_func():print()class TestAdd: # def __init__(self): 在 pytest 中不可以使用__init__方法 # self.cc 12345 pytest.mark.api def test_str(self):res add(1, 2)assert res 12def test_int(self):r…...

视觉slam十四讲实践部分记录——ch2、ch3

ch2 一、使用g++编译.cpp为可执行文件并运行(P30) g++ helloSLAM.cpp ./a.out运行 二、使用cmake编译 mkdir build cd build cmake .. makeCMakeCache.txt 文件仍然指向旧的目录。这表明在源代码目录中可能还存在旧的 CMakeCache.txt 文件,或者在构建过程中仍然引用了旧的路…...

关于easyexcel动态下拉选问题处理

前些日子突然碰到一个问题&#xff0c;说是客户的导入文件模版想支持部分导入内容的下拉选&#xff0c;于是我就找了easyexcel官网寻找解决方案&#xff0c;并没有找到合适的方案&#xff0c;没办法只能自己动手并分享出来&#xff0c;针对Java生成Excel下拉菜单时因选项过多导…...

Linux部署私有文件管理系统MinIO

最近需要用到一个文件管理服务&#xff0c;但是又不想花钱&#xff0c;所以就想着自己搭建一个&#xff0c;刚好我们用的一个开源框架已经集成了MinIO&#xff0c;所以就选了这个 我这边对文件服务性能要求不是太高&#xff0c;单机版就可以 安装非常简单&#xff0c;几个命令就…...

C++ 类基础:封装、继承、多态与多线程模板实现

前言 C 是一门强大的面向对象编程语言&#xff0c;而类&#xff08;Class&#xff09;作为其核心特性之一&#xff0c;是理解和使用 C 的关键。本文将深入探讨 C 类的基本特性&#xff0c;包括封装、继承和多态&#xff0c;同时讨论类中的权限控制&#xff0c;并展示如何使用类…...

Qt的学习(二)

1. 创建Hello Word 两种方式&#xff0c;实现helloworld&#xff1a; 1.通过图形化的方式&#xff0c;在界面上创建出一个控件&#xff0c;显示helloworld 2.通过纯代码的方式&#xff0c;通过编写代码&#xff0c;在界面上创建控件&#xff0c; 显示hello world&#xff1b; …...

深入浅出JavaScript中的ArrayBuffer:二进制数据的“瑞士军刀”

深入浅出JavaScript中的ArrayBuffer&#xff1a;二进制数据的“瑞士军刀” 在JavaScript中&#xff0c;我们经常需要处理文本、数组、对象等数据类型。但当我们需要处理文件上传、图像处理、网络通信等场景时&#xff0c;单纯依赖字符串或数组就显得力不从心了。这时&#xff…...