【大数据哔哔集20210108】Spark Shuffle 和 Hadoop Shuffle有什么异同？

Shuffle的本意是洗牌、混洗的意思，把一组有规则的数据尽量打乱成无规则的数据。而在MapReduce中，Shuffle更像是洗牌的逆过程，指的是将map端的无规则输出按指定的规则“打乱”成具有一定规则的数据，以便reduce端接收处理。其在MapReduce中所处的工作阶段是map输出后到reduce接收前，具体可以分为map端和reduce端前后两个部分。

在Shuffle之前，也就是在map阶段，MapReduce会对要处理的数据进行分片（split）操作，为每一个分片分配一个MapTask任务。接下来map会对每一个分片中的每一行数据进行处理得到键值对（key,value）此时得到的键值对又叫做“中间结果”。此后便进入reduce阶段，由此可以看出Shuffle阶段的作用是处理“中间结果”。

由于Shuffle涉及到了磁盘的读写和网络的传输，因此Shuffle性能的高低直接影响到了整个程序的运行效率。

我们先放一张图，下图说明了二者shuffle的主要区别：

MapReduce Shuffle

我们在《大数据哔哔集20210107》中详细讲解过MapReduce的shuffle过程：

map阶段

  1. 在map task执行时，它的输入数据来源于HDFS的block，当然在MapReduce概念中，map task只读取split。Split与block的对应关系可能是多对一，默认是一对一。

  2. 在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对： key是“hello”， value是数值1。因为当前map端只做加1的操作，在reduce task里才去合并结果集。这个job有3个reduce task，到底当前的“hello”应该交由哪个reduce去做呢，是需要现在决定的。

  • 分区（partition）
    MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。
    一个split被分成了3个partition。

  • 排序sort
    在spill写入之前，会先进行二次排序，首先根据数据所属的partition进行排序，然后每个partition中的数据再按key来排序。partition的目是将记录划分到不同的Reducer上去，以期望能够达到负载均衡，以后的Reducer就会根据partition来读取自己对应的数据。接着运行combiner(如果设置了的话)，combiner的本质也是一个Reducer，其目的是对将要写入到磁盘上的文件先进行一次处理，这样，写入到磁盘的数据量就会减少。

  • 溢写（spill）
    Map端会处理输入数据并产生中间结果，这个中间结果会写到本地磁盘，而不是HDFS。每个Map的输出会先写到内存缓冲区中， 缓冲区的作用是批量收集map结果，减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。 当写入的数据达到设定的阈值时，系统将会启动一个线程将缓冲区的数据写到磁盘，这个过程叫做spill。
    这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8。
    将数据写到本地磁盘产生spill文件(spill文件保存在{mapred.local.dir}指定的目录中，MapReduce任务结束后就会被删除)。

  • 合并（merge）
    每个Map任务可能产生多个spill文件，在每个Map任务完成前，会通过多路归并算法将这些spill文件归并成一个文件。这个操作就叫merge(spill文件保存在{mapred.local.dir}指定的目录中，Map任务结束后就会被删除)。一个map最终会溢写一个文件。
    至此，Map的shuffle过程就结束了。

Reduce阶段

Reduce端的shuffle主要包括三个阶段，copy、sort(merge)和reduce。

  • copy
    首先要将Map端产生的输出文件拷贝到Reduce端，但每个Reducer如何知道自己应该处理哪些数据呢？因为Map端进行partition的时候，实际上就相当于指定了每个Reducer要处理的数据(partition就对应了Reducer)，所以Reducer在拷贝数据的时候只需拷贝与自己对应的partition中的数据即可。每个Reducer会处理一个或者多个partition，但需要先将自己对应的partition中的数据从每个Map的输出结果中拷贝过来。

  • merge
    Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size设置，因为Shuffle阶段Reducer不运行，所以应该把绝大部分的内存都给Shuffle用。

这里需要强调的是：

merge阶段，也称为sort阶段，因为这个阶段的主要工作是执行了归并排序。从Map端拷贝到Reduce端的数据都是有序的，所以很适合归并排序。

merge有三种形式：1)内存到内存 2)内存到磁盘 3)磁盘到磁盘。默认情况下第一种形式不启用，让人比较困惑，是吧。

当copy到内存中的数据量到达一定阈值，就启动内存到磁盘的merge，即第二种merge方式，与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。这种merge方式一直在运行，直到没有map端的数据时才结束。然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。

  • reduce
    不断地merge后，最后会生成一个“最终文件”。为什么加引号？因为这个文件可能存在于磁盘上，也可能存在于内存中。对我们来说，当然希望它存放于内存中，直接作为Reducer的输入，但默认情况下，这个文件是存放于磁盘中的。至于怎样才能让这个文件出现在内存中，参见性能优化篇。
    然后就是Reducer执行，在这个过程中产生了最终的输出结果，并将其写到HDFS上。

Spark Shuffle

我们在之前的文章《Spark性能优化总结》中提到过，Spark Shuffle 的原理和演进过程。

Spark在DAG阶段以宽依赖shuffle为界，划分stage，上游stage做map task，每个map task将计算结果数据分成多份，每一份对应到下游stage的每个partition中，并将其临时写到磁盘，该过程叫做shuffle write。

下游stage做reduce task，每个reduce task通过网络拉取上游stage中所有map task的指定分区结果数据，该过程叫做shuffle read，最后完成reduce的业务逻辑。

下图中，上游stage有3个map task，下游stage有4个reduce task，那么这3个map task中每个map task都会产生4份数据。而4个reduce task中的每个reduce task都会拉取上游3个map task对应的那份数据。

Spark Shuffle演进

  • < 0.8 hashBasedShuffle
    每个map端的task为每个reduce端的partition/task生成一个文件，通常会产生大量的文件，伴随大量的随机磁盘IO操作与大量的内存开销M * R

  • 0.8.1 引入文件合并File Consolidation机制
    每个executor为每个reduce端的partition生成一个文件E*R

  • 0.9 引入External AppendOnlyMap
    combine时可以将数据spill到磁盘，然后通过堆排序merge

  • 1.1 引入sortBasedShuffle
    每个map task不会为每个reducer task生成一个单独的文件，而是会将所有的结果写到一个文件里，同时会生成一个index文件，reducer可以通过这个index文件取得它需要处理的数据M

  • 1.4 引入Tungsten-Sort Based Shuffle
    亦称unsafeShuffle，将数据记录用序列化的二进制方式存储，把排序转化成指针数组的排序，引入堆外内存空间和新的内存管理模型

  • 1.6 Tungsten-sort并入Sort Based Shuffle
    由SortShuffleManager自动判断选择最佳Shuffle方式，如果检测到满足Tungsten-sort条件会自动采用Tungsten-sort Based Shuffle，否则采用Sort Based Shuffle

  • 2.0 hashBasedShuffle退出历史舞台
    从此Spark只有sortBasedShuffle

总结

到此为止，我们已经把二者的原理完完整整的讲了一遍。最后，总结引用ITStar总结过的二者的不同精简要点版本：

Hadoop Shuffle:通过Map端处理的数据到Reduce端的中间的过程就是Shuffle.

Spark Shuffle:在DAG调度过程中,stage阶段的划分是根据shuffle过程,也就是存在ShuffleDependency宽窄依赖的时候,需要进行shuffle,(这时候会将作业Job划分成多个stage;并且在划分stage的时候,构建shuffleDependency的时候进行shuffle注册,获取后续数据读取所需要的shuffleHandle),最终每一个Job提交后都会生成一个ResultStage和若干个ShuffleMapStage.

shuffle过程排序次数不同

Hadoop Shuffle过程中总共发生3次排序,详细分别如下:

第一次排序行为:在map阶段,由环形缓冲区溢出到磁盘上时,落地磁盘的文件会按照key进行分区和排序,属于分区内有序,排序算法为快速排序.

第二次排序行为:在map阶段,对溢出的文件进行combiner合并过程中,需要对溢出的小文件进行归档排序,合并,排序算法为归并排序.

第三次排序行为:在map阶段,reduce task将不同map task端文件拉取到同一个reduce分区后,对文件进行合并,排序,排序算法为归并排序.

spark shuffle过程在满足shuffle manager为sortshuffleManager,且运行模式为普通模式的情况下才会发生排序行为,排序行为发生在数据结构中保存数据内存达到阀值,再溢出磁盘文件之前会对内存数据结构中数据进行排序;

spark中sorted-Based Shuffle在Mapper端是进行排序的,包括partition的排序和每个partition内部元素进行排序,但是在Reducer端没有进行排序,所有job的结果默认情况下不是排序的.Sprted-Based Shuffle 采用 Tim-Sort排序算法,好处是可以极为高效的使用Mapper端的排序成果完成全局排序.

shuffle 逻辑流划分

Hadoop是基于文件的数据结构，Spark是基于RDD的数据结构,计算性能要比Hadoop要高。

Shuffle Fetch后数据存放位置

Hadoopreduce 端将 map task 的文件拉取到同一个reduce分区,是将文件进行归并排序,合并,将文件直接保存在磁盘上。

SparkShuffle Read 拉取来的数据首先肯定是放在Reducer端的内存缓存区中的,现在的实现都是内存+磁盘的方式(数据结构使用 ExternalAppendOnlyMap),当然也可以通过Spark.shuffle.spill=false来设置只能使用内存.使用ExternalAppendOnlyMap的方式时候如果内存的使用达到一定临界值,会首先尝试在内存中扩大ExternalAppendOnlyMap(内部有实现算法),如果不能扩容的话才会spil到磁盘.

什么时候进行Shuffle Fetch操作

Hadoop Shuffle把数据拉过来之后,然后进行计算，如果用MapReduce求平均值的话,它的算法就会很好实现。Spark Shuffle的过程是边拉取数据边进行Aggregrate操作。

Fetch操作与数据计算粒度

Hadoop的MapReduce是粗粒度的，Hadoop Shuffle Reducer Fetch 到的数据record先暂时被存放到Buffer中,当Buffer快满时才进行combine()操作。Spark的Shuffle Fetch是细粒度的,Reducer是对Map端数据Record边拉取边聚合。

性能优化的角度

Hadoop MapReduce的shuffle方式单一.Spark针对不同类型的操作，不同类型的参数,会使用不同的shuffle write方式;而spark更加全面。

赞 (0)

相关推荐