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

相关文章

PowerShell中15个提升运维效率关键命令实战指南

《PowerShell中15个提升运维效率关键命令实战指南》作为网络安全专业人员的必备技能,PowerShell在系统管理、日志分析、威胁检测和自动化响应方面展现出强大能力,下面我们就来看看15个提升... 目录一、PowerShell在网络安全中的战略价值二、网络安全关键场景命令实战1. 系统安全基线核查

从原理到实战深入理解Java 断言assert

《从原理到实战深入理解Java断言assert》本文深入解析Java断言机制,涵盖语法、工作原理、启用方式及与异常的区别,推荐用于开发阶段的条件检查与状态验证,并强调生产环境应使用参数验证工具类替代... 目录深入理解 Java 断言(assert):从原理到实战引言:为什么需要断言?一、断言基础1.1 语

MyBatisPlus如何优化千万级数据的CRUD

《MyBatisPlus如何优化千万级数据的CRUD》最近负责的一个项目,数据库表量级破千万,每次执行CRUD都像走钢丝,稍有不慎就引起数据库报警,本文就结合这个项目的实战经验,聊聊MyBatisPl... 目录背景一、MyBATis Plus 简介二、千万级数据的挑战三、优化 CRUD 的关键策略1. 查

MySQL中的表连接原理分析

《MySQL中的表连接原理分析》:本文主要介绍MySQL中的表连接原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1、背景2、环境3、表连接原理【1】驱动表和被驱动表【2】内连接【3】外连接【4编程】嵌套循环连接【5】join buffer4、总结1、背景

深度解析Spring AOP @Aspect 原理、实战与最佳实践教程

《深度解析SpringAOP@Aspect原理、实战与最佳实践教程》文章系统讲解了SpringAOP核心概念、实现方式及原理,涵盖横切关注点分离、代理机制(JDK/CGLIB)、切入点类型、性能... 目录1. @ASPect 核心概念1.1 AOP 编程范式1.2 @Aspect 关键特性2. 完整代码实

Java Stream的distinct去重原理分析

《JavaStream的distinct去重原理分析》Javastream中的distinct方法用于去除流中的重复元素,它返回一个包含过滤后唯一元素的新流,该方法会根据元素的hashcode和eq... 目录一、distinct 的基础用法与核心特性二、distinct 的底层实现原理1. 顺序流中的去重

Spring @Scheduled注解及工作原理

《Spring@Scheduled注解及工作原理》Spring的@Scheduled注解用于标记定时任务,无需额外库,需配置@EnableScheduling,设置fixedRate、fixedDe... 目录1.@Scheduled注解定义2.配置 @Scheduled2.1 开启定时任务支持2.2 创建

Spring Boot 实现 IP 限流的原理、实践与利弊解析

《SpringBoot实现IP限流的原理、实践与利弊解析》在SpringBoot中实现IP限流是一种简单而有效的方式来保障系统的稳定性和可用性,本文给大家介绍SpringBoot实现IP限... 目录一、引言二、IP 限流原理2.1 令牌桶算法2.2 漏桶算法三、使用场景3.1 防止恶意攻击3.2 控制资源

Python中使用uv创建环境及原理举例详解

《Python中使用uv创建环境及原理举例详解》uv是Astral团队开发的高性能Python工具,整合包管理、虚拟环境、Python版本控制等功能,:本文主要介绍Python中使用uv创建环境及... 目录一、uv工具简介核心特点:二、安装uv1. 通过pip安装2. 通过脚本安装验证安装:配置镜像源(可

Mysql的主从同步/复制的原理分析

《Mysql的主从同步/复制的原理分析》:本文主要介绍Mysql的主从同步/复制的原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录为什么要主从同步?mysql主从同步架构有哪些?Mysql主从复制的原理/整体流程级联复制架构为什么好?Mysql主从复制注意