Spark Shuffle原理详解

Posted <一蓑烟雨任平生>

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark Shuffle原理详解相关的知识,希望对你有一定的参考价值。

(1)Shuffle概述

Shuffle 就是对数据进行重组,是把一组无规则的数据尽量转换成一组具有一定规则的数 据。由于分布式计算的特性和要求,在实现细节上更加繁琐和复杂。

在 MapReduce 框架,Shuffle 是连接 Map 和 Reduce 之间的桥梁,Map 阶段通过 shuffle 读取 数据并输出到对应的 Reduce。而 Reduce 阶段负责从 Map 端拉取数据并进行计算。在整个 shuffle 过程中,往往伴随着大量的磁盘和网络 I/O。所以 shuffle 性能的高低也直接决定 了整个程序的性能高低。下图为 Hadoop Shuffle 过程。

Spark 也有自己的 shuffle 实现过程。在 DAG 调度的过程中,Stage 阶段的划分是根据是否 有 shuffle 过程,也就是存在 ShuffleDependency 宽依赖的时候,需要进行 shuffle。 并且在划分 Stage 并构建 ShuffleDependency 的时候进 行 shuffle 注册,获取后续数据读取所需要的 ShuffleHandle, 最终每一个 job 提交后都会 生成一个 ResultStage 和若干个 ShuffleMapStage,其中 ResultStage 表示生成作业的最终 结果所在的 Stage。ResultStage 与 ShuffleMapStage 中的 task 分别对应着 ResultTask 与 ShuffleMapTask。一个作业,除了最终的 ResultStage 外,其他若干 ShuffleMapStage 中各 个 ShuffleMapTask 都需要将最终的数据根据相应的 Partitioner 对数据进行分组,然后持 久化分区的数据。

(2)Hash Shuffle机制

(2.1)Hash Shuffle概述

在 spark-1.6 版本之前,一直使用 HashShuffle,在 spark-1.6 版本之后使用 Sort Shuffle,因为 HashShuffle 存在的不足所以就替换了 HashShuffle。

我们知道,Spark 的运行主要分为 2 部分:一部分是驱动程序,其核心是 SparkContext。另 一部分是 Worker 节点上 Task,它是运行实际任务的。程序运行的时候,Driver 和 Executor 进程相互交互,Driver 会分配 Task 到 Executor,也就是 Driver 跟 Executor 会进行网络 传输。另外,当前 Task 要抓取其他上游的 Task 的数据结果,所以这个过程中就不断的产 生网络结果。其中下一个 Stage 向上一个 Stage 要数据这个过程,我们就称之为 Shuffle。

(2.2)没有优化之前的Hash Shuffle机制


HashShuffle没有优化之前的细节过程:

在 HashShuffle 没有优化之前,每一个 ShufflleMapTask 会为每一个 ReduceTask 创建一个bucket 缓存,并且会为每一个 bucket 创建一个文件。这个 bucket 存放的数据就是经过 Partitioner 操作(默认是 HashPartitioner)之后找到对应的 bucket 然后放进去,最后将数 据刷新 bucket 缓存的数据到磁盘上,即对应的 block file。

然 后 ShuffleMapTask 将 输 出 作 为 MapStatus 发 送 到 DAGScheduler 的 MapOutputTrackerMaster,每一个 MapStatus 包含了每一个 ResultTask 要拉取的数据的位 置和大小。

接下来 ResultTask 去利用 BlockStoreShuffleFetcher 向 MapOutputTrackerMaster 获取 MapStatus,看哪一份数据是属于自己的,然后底层通过 BlockManager 将数据拉取过来。

拉取过来的数据会组成一个内部的 ShuffleRDD,优先放入内存,内存不够用则放入磁盘, 然后 ResulTask 开始进行聚合,最后生成我们希望获取的那个 MapPartitionRDD。

自己的话总结:

每一个上游ShufflleMapTask 根据下游 ReduceTask数量,产生对应多个的bucket内存,这个bucket存放的数据是经过Partition操作(默认是Hashpartition)之后找到对应的 bucket 然后放进去,bucket内存大小默认是32k,最后将bucket缓存的数据溢写到磁盘,即为对应的block file。接下来Reduce Task底层通过 BlockManager 将数据拉取过来。拉取过来的数据会组成一个内部的 ShuffleRDD,优先放入内存,内存不够用则放入磁盘。

没有优化的Hash Shuffle的缺点

如上图所示:在这里有 1 个 worker, 2 个 executor, 每一个 executor 运行 2 个 ShuffleMapTask, 有三个 ReduceTask, 计算方式为:executor 数量每个 executor 的 ShuffleMapTask 数量ReduceTask 数量。所以总共就有 2 * 2 * 3=12 个 bucket 以及对应 12 个 block file(分区文件)。

  • 如果数据量较大,将会生成 M*R 个小文件,比如 ShuffleMapTask 有 100 个,ResultTask 有 100 个,这就会产生 100 * 100=10000 个小文件
  • bucket 缓存很重要,需要将 ShuffleMapTask 所有数据都写入 bucket,然后再刷到磁盘。 那么如果 Map 端数据过多,这就很容易造成内存溢出。尽管后面有优化,bucket 写入的数 据达到刷新到磁盘的阀值之后,就会将数据一点一点的刷新到磁盘,但是这样磁盘 I/O 就多 了。

(2.3)优化后的Hash Shuffle机制


HashShuffle优化之后的细节过程:
每一个 Executor 进程根据核数,决定 Task 的并发数量,比如 executor 核数是 2,那就可 以并发运行两个 task,如果是一个则只能运行一个 task。

假设 executor 核数是 1,ShuffleMapTask 数量是 M,那么它依然会根据 ResultTask 的数量 R, 创建 R 个 bucket 缓存,然后对 key 进行 hash,数据进入不同的 bucket 中,每一个 bucket 对应着一个 block file,用于刷新 bucket 缓存里的数据。

然后下一个 task 运行的时候,就不会再创建新的 bucket 和 block file,而是复用之前的 task 已经创建好的 bucket 和 block file。即所谓同一个 Executor 进程里所有 Task 都会把 相同的 key 放入相同的 bucket 缓冲区中。

这样的话, 生成文件的数量就是(本地 worker 的所有 executor 对应的 cores 的总数 *ResultTask 数量)如上图所示,即 2 * 3 = 6 个文件,每一个 Executor 的 shuffleMapTask 数量 100,ReduceTask 数量即为 100。

接下来举例比较一下,未优化的 HashShuffle 的文件数是 2 * 100 * 100 =20000,优化之后的 数量是 2*100 = 200 文件,相当于少了 100 倍。

自己的话总结:

每一个 Executor 进程根据核数,决定 Task 的并发数量,如果executor 核数是1,则只能运行一个task。ShuffleMapTask 会根据 ResultTask 的数量,产生对应多个的bucket内存,然后对 key 进行 hash分区,数据进入不同的 bucket 中,每一个 bucket 对应着一个 block file,用于刷新 bucket缓存里的数据。然后下一个 task 运行的时候,就不会再创建新的 bucket 和 block file,而是复用之前的 task 已经创建好的 bucket 和 block file。即所谓同一个 Executor 进程里所有 Task 都会把 相同的 key放入相同的 bucket 缓冲区中。

优化过的Hash Shuffle的缺点
如果 Reducer 端的并行任务或者是数据分片过多的话则 Core * Reducer Task 依旧 过大,也会产生很多小文件。

(3)Sort Shuffle机制

为了缓解 Shuffle 过程产生文件数过多和 Writer 缓存开销过大的问题,spark 引入了类似 于 hadoop Map-Reduce 的 shuffle 机制。该机制每一个 ShuffleMapTask 不会为后续的任务 创建单独的文件,而是会将所有的 Task 结果写入同一个文件,并且对应生成一个索引文件。 以前的数据是放在内存缓存中,等到缓存读取完数据后再刷到磁盘,现在为了减少内存的使 用,在内存不够用的时候,可以将输出溢写到磁盘。结束的时候,再将这些不同的文件联合 内存(缓存)的数据一起进行归并,从而减少内存的使用量。一方面文件数量显著减少,另 一方面减少 Writer 缓存所占用的内存大小,而且同时避免 GC 的风险和频率。

Sort Shuffle原理: 普通的SortShuffle

在普通模式下,数据会先写入一个内存数据结构中,此时根据不同的shuffle算子,可以选用不同的数据结构。如果是由聚合操作的shuffle算子,就是用map的数据结构(边聚合边写入内存),如果是join的算子,就使用array的数据结构(直接写入内存)。接着,每写一条数据进入内存数据结构之后,就会判断是否达到了某个临界值,如果达到了临界值的话,就会尝试的将内存数据结构中的数据溢写到磁盘,然后清空内存数据结构。

在溢写到磁盘文件之前,会先根据key对内存数据结构中已有的数据进行排序,排序之后,会分批将数据写入磁盘文件。默认的batch数量是10000条,也就是说,排序好的数据,会以每批次1万条数据的形式分批写入磁盘文件,写入磁盘文件是通过Java的BufferedOutputStream实现的。BufferedOutputStream是Java的缓冲输出流,首先会将数据缓冲在内存中,当内存缓冲满溢之后再一次写入磁盘文件中,这样可以减少磁盘IO次数,提升性能。

此时task将所有数据写入内存数据结构的过程中,会发生多次磁盘溢写,会产生多个临时文件,最后会将之前所有的临时文件都进行合并,最后会合并成为一个大文件。最终只剩下两个文件,一个是合并之后的数据文件,一个是索引文件(标识了下游各个task的数据在文件中的start offset与end offset)。最终再由下游的task根据索引文件读取相应的数据文件。

Sort Shuffle原理: bypass运行机制

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

bypass机制与普通SortShuffleManager运行机制的不同在于:

a、磁盘写机制不同;
b、不会进行排序。

也就是说,启用该机制的最大好处在于,shuffle write过程中,不需要进行数据的排序操作,也就节省掉了这部分的性能开销。

自己的话总结:
普通的SortShuffle机制

在普通模式下,数据会先写入一个内存数据结构中,如果是由聚合操作的shuffle算子用map数据结构,如果是join算子就用Array数据结构。在写入的过程中如果达到了临界值,就会将内存数据结构中的数据溢写到磁盘,然后清空内存数据结构。在溢写磁盘之前,会先根据key对内存数据结构中的数据进行排序,排序好的数据,会以每批次1万条数据的形式分批写入磁盘文件。在task将所有数据写入内存数据结构的过程中,会发生多次磁盘溢写,会产生多个临时文件,最后会将之前所有的临时文件都进行合并,最后会合并成为一个大文件。最终只剩下两个文件,一个是合并之后的数据文件,一个是索引文件,最终再由下游的task根据索引文件读取相应的数据文件。

bypass运行机制

bypass的就是不排序,还是用hash去为key分磁盘文件,分完之后再合并,形成一个索引文件和一个合并后的key hash文件。省掉了排序的性能。

Sort Shuffle 有几种不同的策略:

  • BypassMergeSortShuffleWriter(Bypass 机制)
  • SortShuffleWriter(普通机制)
  • UnsafeShuffleWriter

对于 BypassMergeSortShuffleWriter,使用这个模式的特点为:

  • 主要用于处理不需要排序和聚合的 Shuffle 操作,所以数据是直接写入文件,数据量较大 的时候,网络 I/O 和内存负担较重。
  • 主要适合处理 Reducer 任务数量比较少的情况。
  • 将每一个分区写入一个单独的文件,最后将这些文件合并,减少文件数量。但是这种方式 需要并发打开多个文件,对内存消耗比较大。

因为 BypassMergeSortShuffleWriter 这种方式比 SortShuffleWriter 更快,所以如果在 Reducer 数 量 不 大 , 又 不 需 要 在 map 端 聚 合 和 排 序 , 而 且 Reducer 的 数 目 小 于 spark.shuffle.sort.bypassMergeThreshold 指定的阀值(默认 200)时,就是用的是这种 方式(即启用条件)。

对于 SortShuffleWriter(普通机制),使用这个模式的特点为:

  • 比较适合数据量很大的场景或者集群规模很大。
  • 引入了外部排序器,可以支持在 Map 端进行本地聚合或者不聚合。
  • 如果外部排序器 enable 了 spill 功能,如果内存不够,可以先将输出溢写到本地磁盘, 最后将内存结果和本地磁盘的溢写文件进行合并。

另外,这个 Sort-Based Shuffle 跟 Executor 核数没有关系,即跟并发度没有关系,它是每 一个 ShuffleMapTask 都会产生一个 data 文件和 index 文件, 所谓合并也只是将该 ShuffleMapTask 的各个 partition 对应的分区文件合并到 data 文件而已。所以这个就需要 和 Hash-BasedShuffle 的 consolidation 机制区别开来。

(4)Spark Shuffle调优

  • 调节map端缓冲区大小

    • map端缓冲的默认配置是32KB
    • val conf = new SparkConf().set("spark.shuffle.file.buffer", "64")
  • 调节reduce端拉取数据缓冲区大小

    • 如果内存资源较为充足,适当增加拉取数据缓冲区的大小,可以减少拉取数据的次数,也就可以减少网络传输的次数,进而提升性能
    • reduce端数据拉取缓冲区大小默认48MB
    • val conf = new SparkConf().set("spark.reducer.maxSizeInFlight", "96")
  • 调节reduce端拉取数据重试次数

    • reduce端拉取数据重试次数默认为3
    • val conf = new SparkConf().set("spark.shuffle.io.maxRetries", "6")
  • 调节reduce端拉取数据等待间隔

    • reduce端拉取数据等待间隔默认为5s
    • val conf = new SparkConf().set("spark.shuffle.io.retryWait", "60s")
  • 调节SortShuffle排序操作阈值

    • SortShuffleManager排序操作阈值默认为200
    • val conf = new SparkConf().set("spark.shuffle.sort.bypassMergeThreshold", "400")

以上内容仅供参考学习,如有侵权请联系我删除!
如果这篇文章对您有帮助,左下角的大拇指就是对博主最大的鼓励。
您的鼓励就是博主最大的动力!

以上是关于Spark Shuffle原理详解的主要内容,如果未能解决你的问题,请参考以下文章

Spark中的Spark Shuffle详解

Spark的两种核心Shuffle详解

Spark的两种核心Shuffle详解

Mr与spark的shuffle过程详解及对比

Mr与spark的shuffle过程详解及对比

第37课:Spark中Shuffle详解及作业