大数据培训：Spark Shuffle的几种情况
作者：张老师 浏览次数： 2021-04-25 17:19
Spark框架当中，基于RDD的基本数据模型，内部的运行会带来各种Shuffle的问题。从RDD到Dataframe、DataSet，不同的Stage转换，Shuffle不可避免，但是在转换当中，会造成性能消耗。今天的大数据培训分享，我们来讲讲Spark Shuffle的几种情况。
一、shuffle定义
一个spark的RDD有一组固定的分区组成，每个分区有一系列的记录组成。对于由窄依赖变换（例如map和filter）返回的RDD，会延续父RDD的分区信息，以pipeline的形式计算。每个对象仅依赖于父RDD中的单个对象。诸如coalesce之类的操作可能导致任务处理多个输入分区，但转换仍然被认为是窄依赖的，因为一个父RDD的分区只会被一个子RDD分区继承。
Spark还支持宽依赖的转换，例如groupByKey和reduceByKey。在这些依赖项中，计算单个分区中的记录所需的数据可以来自于父数据集的许多分区中。要执行这些转换，具有相同key的所有元组必须最终位于同一分区中，由同一任务处理。为了满足这一要求，Spark产生一个shuffle，它在集群内部传输数据，并产生一个带有一组新分区的新stage。
在每个stage的边界，父stage的task会将数据写入磁盘，子stage的task会将数据通过网络读取。由于它们会导致很高的磁盘和网络IO，所以shuffle代价相当高，应该尽量避免。父stage的数据分区往往和子stage的分区数不同。触发shuffle的操作算子往往可以指定分区数的，也即是numPartitions代表下个stage会有多少个分区。就像mr任务中reducer的数据是非常重要的一个参数一样，shuffle的时候指定分区数也将在很大程度上决定一个应用程序的性能。
二、优化shuffle
通常情况可以选择使用产生相同结果的action和transform相互替换。但是并不是产生相同结果的算子就会有相同的性能。通常避免常见的陷阱并选择正确的算子可以显著提高应用程序的性能。
当选择转换操作的时候，应最小化shuffle次数和shuffle的数据量。shuffle是非常消耗性能的操作。所有的shuffle数据都会被写入磁盘，然后通过网络传输。repartition,join,cogroup,和*By或者*ByKey类型的操作都会产生shuffle。我们可以对一下几个操作算子进行优化：
1.groupByKey某些情况下可以被reducebykey代替。
2.reduceByKey某些情况下可以被aggregatebykey代替。
3.flatMap-join-groupBy某些情况下可以被cgroup代替。
三、no shuffle
在某些情况下，前面描述的转换操作不会导致shuffle。当先前的转换操作已经使用了和shuffle相同的分区器分区数据的时候，spark就不会产生shuffle。
举个例子：
rdd1=someRdd.reduceByKey(...)
rdd2=someOtherRdd.reduceByKey(...)
rdd3=rdd1.join(rdd2)
由于使用redcuebykey的时候没有指定分区器，所以都是使用的默认分区器，会导致rdd1和rdd2都采用的是hash分区器。两个reducebykey操作会产生两个shuffle过程。如果，数据集有相同的分区数，执行join操作的时候就不需要进行额外的shuffle。由于数据集的分区相同，因此rdd1的任何单个分区中的key集合只能出现在rdd2的单个分区中。因此，rdd3的任何单个输出分区的内容仅取决于rdd1中单个分区的内容和rdd2中的单个分区，并且不需要第三个shuffle。
四、增加shuffle
有时候需要打破最小化shuffle次数的规则。
当增加并行度的时候，额外的shuffle是有利的。例如，数据中有一些文件是不可分割的，那么该大文件对应的分区就会有大量的记录，而不是说将数据分散到尽可能多的分区内部来使用所有已经申请cpu。在这种情况下，使用reparition重新产生更多的分区数，以满足后面转换算子所需的并行度，这会提升很大性能。
关于大数据培训，Spark Shuffle的几种情况，以上就为大家做了大致的介绍了。Spark Shuffle是Spark程序运行当中需要考虑的一个重要环节，学习当中需要对Shuffle做到全面的理解和掌握。成都加米谷大数据，专业大数据培训机构，大数据开发、数据分析与挖掘，零基础班本月正在招生中，课程大纲及学习资料，可联系客服获取！
一、shuffle定义
一个spark的RDD有一组固定的分区组成，每个分区有一系列的记录组成。对于由窄依赖变换（例如map和filter）返回的RDD，会延续父RDD的分区信息，以pipeline的形式计算。每个对象仅依赖于父RDD中的单个对象。诸如coalesce之类的操作可能导致任务处理多个输入分区，但转换仍然被认为是窄依赖的，因为一个父RDD的分区只会被一个子RDD分区继承。
Spark还支持宽依赖的转换，例如groupByKey和reduceByKey。在这些依赖项中，计算单个分区中的记录所需的数据可以来自于父数据集的许多分区中。要执行这些转换，具有相同key的所有元组必须最终位于同一分区中，由同一任务处理。为了满足这一要求，Spark产生一个shuffle，它在集群内部传输数据，并产生一个带有一组新分区的新stage。
在每个stage的边界，父stage的task会将数据写入磁盘，子stage的task会将数据通过网络读取。由于它们会导致很高的磁盘和网络IO，所以shuffle代价相当高，应该尽量避免。父stage的数据分区往往和子stage的分区数不同。触发shuffle的操作算子往往可以指定分区数的，也即是numPartitions代表下个stage会有多少个分区。就像mr任务中reducer的数据是非常重要的一个参数一样，shuffle的时候指定分区数也将在很大程度上决定一个应用程序的性能。
二、优化shuffle
通常情况可以选择使用产生相同结果的action和transform相互替换。但是并不是产生相同结果的算子就会有相同的性能。通常避免常见的陷阱并选择正确的算子可以显著提高应用程序的性能。
当选择转换操作的时候，应最小化shuffle次数和shuffle的数据量。shuffle是非常消耗性能的操作。所有的shuffle数据都会被写入磁盘，然后通过网络传输。repartition,join,cogroup,和*By或者*ByKey类型的操作都会产生shuffle。我们可以对一下几个操作算子进行优化：
1.groupByKey某些情况下可以被reducebykey代替。
2.reduceByKey某些情况下可以被aggregatebykey代替。
3.flatMap-join-groupBy某些情况下可以被cgroup代替。
三、no shuffle
在某些情况下，前面描述的转换操作不会导致shuffle。当先前的转换操作已经使用了和shuffle相同的分区器分区数据的时候，spark就不会产生shuffle。
举个例子：
rdd1=someRdd.reduceByKey(...)
rdd2=someOtherRdd.reduceByKey(...)
rdd3=rdd1.join(rdd2)
由于使用redcuebykey的时候没有指定分区器，所以都是使用的默认分区器，会导致rdd1和rdd2都采用的是hash分区器。两个reducebykey操作会产生两个shuffle过程。如果，数据集有相同的分区数，执行join操作的时候就不需要进行额外的shuffle。由于数据集的分区相同，因此rdd1的任何单个分区中的key集合只能出现在rdd2的单个分区中。因此，rdd3的任何单个输出分区的内容仅取决于rdd1中单个分区的内容和rdd2中的单个分区，并且不需要第三个shuffle。
四、增加shuffle
有时候需要打破最小化shuffle次数的规则。
当增加并行度的时候，额外的shuffle是有利的。例如，数据中有一些文件是不可分割的，那么该大文件对应的分区就会有大量的记录，而不是说将数据分散到尽可能多的分区内部来使用所有已经申请cpu。在这种情况下，使用reparition重新产生更多的分区数，以满足后面转换算子所需的并行度，这会提升很大性能。
关于大数据培训，Spark Shuffle的几种情况，以上就为大家做了大致的介绍了。Spark Shuffle是Spark程序运行当中需要考虑的一个重要环节，学习当中需要对Shuffle做到全面的理解和掌握。成都加米谷大数据，专业大数据培训机构，大数据开发、数据分析与挖掘，零基础班本月正在招生中，课程大纲及学习资料，可联系客服获取！
