Spark技术内幕:Sort Based Shuffle实现解析

本文涉及的产品
全局流量管理 GTM,标准版 1个月
云解析 DNS,旗舰版 1个月
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
简介:

在Spark 1.2.0中,Spark Core的一个重要的升级就是将默认的Hash Based Shuffle换成了Sort Based Shuffle,即spark.shuffle.manager 从hash换成了sort,对应的实现类分别是org.apache.spark.shuffle.hash.HashShuffleManager和org.apache.spark.shuffle.sort.SortShuffleManager。

这个方式的选择是在org.apache.spark.SparkEnv完成的:

    // Let the user specify short names forshuffle managers
    val shortShuffleMgrNames = Map(
      "hash" ->"org.apache.spark.shuffle.hash.HashShuffleManager",
      "sort" ->"org.apache.spark.shuffle.sort.SortShuffleManager")
    val shuffleMgrName =conf.get("spark.shuffle.manager", "sort") //获得Shuffle Manager的type,sort为默认
    val shuffleMgrClass =shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase, shuffleMgrName)
    val shuffleManager =instantiateClass[ShuffleManager](shuffleMgrClass)

那么Sort BasedShuffle“取代”Hash BasedShuffle作为默认选项的原因是什么?

正如前面提到的,Hashbased shuffle的每个mapper都需要为每个reducer写一个文件,供reducer读取,即需要产生M*R个数量的文件,如果mapper和reducer的数量比较大,产生的文件数会非常多。Hash based shuffle设计的目标之一就是避免不需要的排序(Hadoop Map Reduce被人诟病的地方,很多不需要sort的地方的sort导致了不必要的开销)。但是它在处理超大规模数据集的时候,产生了大量的DiskIO和内存的消耗,这无疑很影响性能。Hash based shuffle也在不断的优化中,正如前面讲到的Spark 0.8.1引入的file consolidation在一定程度上解决了这个问题。为了更好的解决这个问题,Spark 1.1 引入了Sort based shuffle。首先,每个Shuffle Map Task不会为每个Reducer生成一个单独的文件;相反,它会将所有的结果写到一个文件里,同时会生成一个index文件,Reducer可以通过这个index文件取得它需要处理的数据。避免产生大量的文件的直接收益就是节省了内存的使用和顺序Disk IO带来的低延时。节省内存的使用可以减少GC的风险和频率。而减少文件的数量可以避免同时写多个文件对系统带来的压力。

并且从作者ReynoldXin的几乎所有的测试来看,Sortbased shuffle在速度和内存使用方面优于Hashbased shuffle:“sort-basedshuffle has lower memory usage and seems to outperformhash-based in almost allof our testing.”

性能数据:from:https://issues.apache.org/jira/browse/SPARK-3280


Shuffle Map Task会按照key相对应的partition ID进行sort,其中属于同一个partition的key不会sort。因为对于不需要sort的操作来说,这个sort是负收益的;要知道之前Spark刚开始使用Hash based的shuffle而不是sort based就是为了避免Hadoop Map Reduce对于所有计算都会sort的性能损耗。对于那些需要sort的运算,比如sortByKey,这个sort在Spark 1.2.0里还是由reducer完成的。

如果这个过程内存不够用了,那么这些已经sort的内容会被spill到外部存储。然后在结束的时候将这些不同的文件进行merge sort。

为了便于下游的Taskfetch到其需要的partition,这里会生成一个index文件,去记录不同的partition的位置信息。当然了org.apache.spark.storage.BlockManager需要也有响应的实现以实现这种新的寻址方式。


核心实现的逻辑都在类org.apache.spark.shuffle.sort.SortShuffleWriter。下面简要分析一下它的实现:

1)          对于每个partition,创建一个scala.Array存储它所包含的key,value对。每个待处理的key,value对都会插入相应的scala.Array。

2)          如果scala.Array的大小超过阈值,那么需要将这个in memory的数据spill到外部存储。这个文件的开始部分会记录这个partition的ID,这个文件保存了多少个pair等信息。

3)          最后需要将所有spill到外部存储的文件进行mergesort。同时打开的文件不能过多,过多的话会消耗大量的内存,增加OOM或者GC的风险;也不能过少,过少的话就会影响性能,增大计算的延时。一般的话推荐每次同时打开10 – 100个文件。

4)          在生成最后的数据文件时,需要同时生成index索引文件。正如前面提到的,这个索引文件将记录不同partition的range。

当然了,你可能还有个疑问,就是Hash Based Shuffle说白了就是根据key需要写入的org.apache.spark.HashPartitioner,为每个Reducer写入单独的Partition。只不过对于同一个Core启动的Shuffle Map Task,如果选择spark.shuffle.consolidateFiles的话,第二个Shuffle Map Task会把结果append到上一个文件中去。那么sort的逻辑是完全可以整合到Hash Based Shuffle中去,为什么又要重新实现一种Shuffle Writer呢?我认为有以下几点:

  • Shuffle机制是所有类似计算模块的核心机制之一,要进行大的优化的风险非常高;比如一个看似简单的consolidation机制,在0.8.1就引入了,但是到1.2.0还是没有作为默认选项。
  • Hash Based Shuffle如果修改为Sort的逻辑,所谓的改进可能会影响原来已经稳定的Spark应用。比如一个应用在使用Hash Based Shuffle性能是完全符合预期的,那么迁移到Spark 1.2.0后,只需要将配置文件修改以下就可以完成这个无缝的迁移。
  • 作为一个通用的计算平台,你的测试的case永远cover不了所有的场景。那么,还是留给用户去选择吧。
  • Sort的机制还处理不断完善的阶段。比如很有的优化或者功能的改进会不断的完善。因此,期待Sort在以后的版本中更加完善吧。

如果您喜欢 本文,那么请动一下手指支持以下博客之星的评比吧。非常感谢您的投票。每天可以一票哦。


目录
相关文章
|
4月前
|
机器学习/深度学习 SQL 分布式计算
Spark核心原理与应用场景解析:面试经验与必备知识点解析
本文深入探讨Spark核心原理(RDD、DAG、内存计算、容错机制)和生态系统(Spark SQL、MLlib、Streaming),并分析其在大规模数据处理、机器学习及实时流处理中的应用。通过代码示例展示DataFrame操作,帮助读者准备面试,同时强调结合个人经验、行业趋势和技术发展以展现全面的技术实力。
262 0
|
4月前
|
分布式计算 Hadoop 大数据
大数据技术与Python:结合Spark和Hadoop进行分布式计算
【4月更文挑战第12天】本文介绍了大数据技术及其4V特性,阐述了Hadoop和Spark在大数据处理中的作用。Hadoop提供分布式文件系统和MapReduce,Spark则为内存计算提供快速处理能力。通过Python结合Spark和Hadoop,可在分布式环境中进行数据处理和分析。文章详细讲解了如何配置Python环境、安装Spark和Hadoop,以及使用Python编写和提交代码到集群进行计算。掌握这些技能有助于应对大数据挑战。
681 1
|
8天前
|
分布式计算 Java Apache
Apache Spark Streaming技术深度解析
【9月更文挑战第4天】Apache Spark Streaming是Apache Spark生态系统中用于处理实时数据流的一个重要组件。它将输入数据分成小批次(micro-batch),然后利用Spark的批处理引擎进行处理,从而结合了批处理和流处理的优点。这种处理方式使得Spark Streaming既能够保持高吞吐量,又能够处理实时数据流。
27 0
|
1月前
|
分布式计算 安全 OLAP
7倍性能提升|阿里云AnalyticDB Spark向量化能力解析
AnalyticDB Spark如何通过向量化引擎提升性能?
|
2月前
|
分布式计算 大数据 Spark
Spark大数据处理:技术、应用与性能优化(全)PDF书籍推荐分享
《Spark大数据处理:技术、应用与性能优化》深入浅出介绍Spark核心,涵盖部署、实战与性能调优,适合初学者。作者基于微软和IBM经验,解析Spark工作机制,探讨BDAS生态,提供实践案例,助力快速掌握。书中亦讨论性能优化策略。[PDF下载链接](https://zhangfeidezhu.com/?p=347)。![Spark Web UI](https://img-blog.csdnimg.cn/direct/16aaadbb4e13410f8cb2727c3786cc9e.png#pic_center)
93 1
Spark大数据处理:技术、应用与性能优化(全)PDF书籍推荐分享
|
3月前
|
分布式计算 Hadoop 大数据
大数据技术:Hadoop与Spark的对比
【6月更文挑战第15天】**Hadoop与Spark对比摘要** Hadoop是分布式系统基础架构,擅长处理大规模批处理任务,依赖HDFS和MapReduce,具有高可靠性和生态多样性。Spark是快速数据处理引擎,侧重内存计算,提供多语言接口,支持机器学习和流处理,处理速度远超Hadoop,适合实时分析和交互式查询。两者在资源占用和生态系统上有差异,适用于不同应用场景。选择时需依据具体需求。
|
3月前
|
机器学习/深度学习 分布式计算 API
技术好文:Spark机器学习笔记一
技术好文:Spark机器学习笔记一
29 0
|
4月前
|
分布式计算 Hadoop 大数据
探索大数据技术:Hadoop与Spark的奥秘之旅
【5月更文挑战第28天】本文探讨了大数据技术中的Hadoop和Spark,Hadoop作为分布式系统基础架构,通过HDFS和MapReduce处理大规模数据,适用于搜索引擎等场景。Spark是快速数据处理引擎,采用内存计算和DAG模型,适用于实时推荐和机器学习。两者各有优势,未来将继续发展和完善,助力大数据时代的发展。
|
4月前
|
分布式计算 Scala Spark
Spark参数解析之MasterArguments
Spark参数解析之MasterArguments
31 0
|
4月前
|
分布式计算 Spark 索引
Spark学习---day07、Spark内核(Shuffle、任务执行)
Spark学习---day07、Spark内核(源码提交流程、任务执行)

推荐镜像

更多