SparkCore(15):Shuffle原理和优化

2024-05-24 11:38

本文主要是介绍SparkCore(15):Shuffle原理和优化,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

一、总括

Shuffle是进行重新分区的过程,即上游RDD与下游RDD是宽依赖的关系。以下操作可能会引起Shuffle
(1)重新调整分区操作:repartiton,coalesce
(2)*ByKey:groupByKey,reduceByKey
(3)关联操作:join


二、shuffle Manager改进

1-》Spark在1.1以前的版本一直是采用Hash Shuffle的实现的方式
2-》1.1版本时参考Hadoop MapReduce的实现开始引入Sort Shuffle
3-》在1.5版本时开始Tungsten钨丝计划,引入UnSafe Shuffle优化内存及CPU的使用
4-》在1.6中将Tungsten统一到Sort Shuffle中,实现自我感知选择最佳Shuffle方式
http://spark.apache.org/docs/1.6.0/configuration.html

5-》到最近的2.0版本,Hash Shuffle已被删除,所有Shuffle方式全部统一到Sort Shuffle一个实现中。下图是spark shuffle实现的一个版本演进。

三、Shuffle Manager具体分类

1.Spark Shuffle Manager:sort(2.0之后默认)

(1)默认执行过程

-》在该模式下,数据会先写入一个数据结构,reduceByKey 写入 Map,一边通过 Map 局部聚合,一遍写入内存。Join 算子写入 ArrayList 直接写入内存中。然后需要判断是否达到阈值,如果达到就会将内存数据结构的数据写入到磁盘,清空内存数据结构。

-》在溢写磁盘前:先根据 key 进行排序,排序过后的数据,会分批写入到磁盘文件中。默 认批次为 10000 条,数据会以每批一万条写入到磁盘文件。

-》在溢写磁盘过程中:写入磁盘文件通过缓冲区溢写的方式,每次溢写都会产生一个磁盘文件,也就是说一个 Task 过程会产生多个临时文件。

-》最后:在每个 Task 中,将所有的临时文件合并,这就是 merge 过程。此过程将所有临时文件读取出来,一次写入到最终文件。意味着一个 Task 的所有数据都在这一个文件中。同时单独写一份索引文件,标识下游各个Task的数据在文件中的索引,start offset和end offset。

 
(2)bypass SortShuffle

bypass SortShuffle机制触发条件:当task的数量小于200的时候,会自动启动by_pass模式(没有数据排序的操作,就是分区完之后直接写入磁盘)。配置参数:spark.shuffle.sort.bypassMergeThreshold:200(默认是200),执行by_pass过程

此时 task 会为每个 reduce 端的 task 都创建一个临时磁盘文件,并将数据按 key 进行hash 然后根据 key 的 hash 值,将 key 写入对应的磁盘文件之中。当然,写入磁盘文件时也是先写入内存缓冲,缓冲写满之后再溢写到磁盘文件的。最后,同样会将所有临时磁盘文件都合并成一个磁盘文件,并创建一个单独的索引文件。

和hashsort对比:

该过程的磁盘写机制其实跟未经优化的 HashShuffleManager 是一模一样的,因为都要创建数量惊人的磁盘文件,只是在最后会做一个磁盘文件的合并而已。因此少量的最终磁 盘文件,也让该机制相对未经优化的 HashShuffleManager 来说,shuffle read 的性能会更好。
而该机制与普通 SortShuffleManager 运行机制的不同在于:不会进行排序。也就是说,启用该机制的最大好处在于,shuffle write 过程中,不需要进行数据的排序操作,也就节省 掉了这部分的性能开销。

2.Spark Shuffle Manager:hash

这里我们先明确一个假设前提:每个 Executor 只有 1 个 CPU core,也就是说,无论这 个 Executor 上分配多少个 task 线程,同一时间都只能执行一个 task 线程。如果Executor有2个以上Cores,则类似,有部分补充,可以参考:https://www.jianshu.com/p/ef45ca960b5d

(1)使用场景

当应用中的数据不需要进行排序的时候(比如reduceByKey,只要分组就好了,不需要排序),可以直接考虑使用hash 

(2)原始的hash执行过程

3 个 Reducer,从 Task 开始那边各自把自己进行 Hash 计算(分区器:hash/numreduce 取模),分类出 3 个不同的类别,每个 Task 都分成 3 种类别的数据,注意:因为每个executor只有一个core,所以executor中的两个task是先后执行的。想把不同的数据汇聚然后计算出最终的结果,所以 Reducer 会在每个 Task 中把属于自己类别的数据收集过来,汇聚成一个同类别的大集合,每 1 个 Task 输出 3 份本地文件,这里有 4 个Mapper Tasks,所以总共输出了 4 个 Tasks x 3 个分类文件 = 12 个本地小文件。
  
即:最初上游maptask1000,下游reducetask1000,shuffle过程有1000000个小文件。所以,面临的问题:一个task就会有多个缓存,然后会有多个文件写出,当task数据量大,linux文件系统崩溃


(3)优化的hash执行过程:

优化的 HashShuffle 过程就是启用合并机制,合并机制就是复用 buffer,开启合并机制的配置是 spark.shuffle.consolidateFiles。该参数默认值为 false,将其设置为 true 即可开启优化机制。通常来说,如果我们使用 HashShuffleManager,那么都建议开启这个选项。

这里还是有 4 个 Tasks,数据类别还是分成 3 种类型,因为 Hash 算法会根据你的 Key 进行分类,在同一个进程中,无论是有多少过 Task,都会把同样的 Key 放在同一个 Buffer里,然后把 Buffer 中的数据写入以 Core 数量为单位的本地文件中,(一个 Core 只有一种类 型的 Key 的数据),每 1 个 Task 所在的进程中,分别写入共同进程中的 3 份本地文件,这里有 4 个 Mapper Tasks,所以总共输出是 2 个 Cores x 3 个分类文件 = 6 个本地小文件。
 
    

    同一个executor中多个task中的buffer先合并,然后再写出文件。

    优化效果,即为原2个executor,每个2个map task,对应3个reduce task,则总共会有2(executor)*3(reduce task)=6个小文件,而不是2*3*3=18个。
    当使用hash shuffle manager的时候(当分区数比较多的),需要将参数:spark.shuffle.consolidateFiles设置为true,表示开启文件合并功能。见:http://spark.apache.org/docs/1.5.2/configuration.html,在1.6+以后已经没有这个参数,如下图

 

四、Shuffle优化

1. 开启shuffle压缩机制

(1)使用场景:磁盘IO使用很满,则shuffle过程使用压缩,通过cpu资源换取整个应用效率
(2)配置参数:spark.shuffle.compress默认true启动。
 
(3)配置参数:spark.shuffle.spill.compress,如果内存不够用,溢写到磁盘, 

 
(4)默认压缩格式:spark.io.compression.codec,1.6前是snappy,2.0后是lz4.

五、参考

1. Spark Shuffle原理及相关调优
https://blog.csdn.net/u010886217/article/details/83409322

2.Spark Shuffle之Hash Shuffle

https://www.jianshu.com/p/ef45ca960b5d

 

这篇关于SparkCore(15):Shuffle原理和优化的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/998306

相关文章

Vue3 的 shallowRef 和 shallowReactive:优化性能

大家对 Vue3 的 ref 和 reactive 都很熟悉,那么对 shallowRef 和 shallowReactive 是否了解呢? 在编程和数据结构中,“shallow”(浅层)通常指对数据结构的最外层进行操作,而不递归地处理其内部或嵌套的数据。这种处理方式关注的是数据结构的第一层属性或元素,而忽略更深层次的嵌套内容。 1. 浅层与深层的对比 1.1 浅层(Shallow) 定义

Ilya-AI分享的他在OpenAI学习到的15个提示工程技巧

Ilya(不是本人,claude AI)在社交媒体上分享了他在OpenAI学习到的15个Prompt撰写技巧。 以下是详细的内容: 提示精确化:在编写提示时,力求表达清晰准确。清楚地阐述任务需求和概念定义至关重要。例:不用"分析文本",而用"判断这段话的情感倾向:积极、消极还是中性"。 快速迭代:善于快速连续调整提示。熟练的提示工程师能够灵活地进行多轮优化。例:从"总结文章"到"用

这15个Vue指令,让你的项目开发爽到爆

1. V-Hotkey 仓库地址: github.com/Dafrok/v-ho… Demo: 戳这里 https://dafrok.github.io/v-hotkey 安装: npm install --save v-hotkey 这个指令可以给组件绑定一个或多个快捷键。你想要通过按下 Escape 键后隐藏某个组件,按住 Control 和回车键再显示它吗?小菜一碟: <template

HDFS—存储优化(纠删码)

纠删码原理 HDFS 默认情况下,一个文件有3个副本,这样提高了数据的可靠性,但也带来了2倍的冗余开销。 Hadoop3.x 引入了纠删码,采用计算的方式,可以节省约50%左右的存储空间。 此种方式节约了空间,但是会增加 cpu 的计算。 纠删码策略是给具体一个路径设置。所有往此路径下存储的文件,都会执行此策略。 默认只开启对 RS-6-3-1024k

深入探索协同过滤:从原理到推荐模块案例

文章目录 前言一、协同过滤1. 基于用户的协同过滤(UserCF)2. 基于物品的协同过滤(ItemCF)3. 相似度计算方法 二、相似度计算方法1. 欧氏距离2. 皮尔逊相关系数3. 杰卡德相似系数4. 余弦相似度 三、推荐模块案例1.基于文章的协同过滤推荐功能2.基于用户的协同过滤推荐功能 前言     在信息过载的时代,推荐系统成为连接用户与内容的桥梁。本文聚焦于

使用opencv优化图片(画面变清晰)

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化,使其看起来更清晰,同时保持尺寸不变,通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多,主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器:相机传

hdu4407(容斥原理)

题意:给一串数字1,2,......n,两个操作:1、修改第k个数字,2、查询区间[l,r]中与n互质的数之和。 解题思路:咱一看,像线段树,但是如果用线段树做,那么每个区间一定要记录所有的素因子,这样会超内存。然后我就做不来了。后来看了题解,原来是用容斥原理来做的。还记得这道题目吗?求区间[1,r]中与p互质的数的个数,如果不会的话就先去做那题吧。现在这题是求区间[l,r]中与n互质的数的和

MySQL高性能优化规范

前言:      笔者最近上班途中突然想丰富下自己的数据库优化技能。于是在查阅了多篇文章后,总结出了这篇! 数据库命令规范 所有数据库对象名称必须使用小写字母并用下划线分割 所有数据库对象名称禁止使用mysql保留关键字(如果表名中包含关键字查询时,需要将其用单引号括起来) 数据库对象的命名要能做到见名识意,并且最后不要超过32个字符 临时库表必须以tmp_为前缀并以日期为后缀,备份

SWAP作物生长模型安装教程、数据制备、敏感性分析、气候变化影响、R模型敏感性分析与贝叶斯优化、Fortran源代码分析、气候数据降尺度与变化影响分析

查看原文>>>全流程SWAP农业模型数据制备、敏感性分析及气候变化影响实践技术应用 SWAP模型是由荷兰瓦赫宁根大学开发的先进农作物模型,它综合考虑了土壤-水分-大气以及植被间的相互作用;是一种描述作物生长过程的一种机理性作物生长模型。它不但运用Richard方程,使其能够精确的模拟土壤中水分的运动,而且耦合了WOFOST作物模型使作物的生长描述更为科学。 本文让更多的科研人员和农业工作者

hdu4407容斥原理

题意: 有一个元素为 1~n 的数列{An},有2种操作(1000次): 1、求某段区间 [a,b] 中与 p 互质的数的和。 2、将数列中某个位置元素的值改变。 import java.io.BufferedInputStream;import java.io.BufferedReader;import java.io.IOException;import java.io.Inpu