Spark 2.0 中已经移除 Hash Based Shuffle,但作为曾经的默认 Shuffle 机制,还是值得进行分析
Spark 最开始只有 Hash Based Shuffle,因为在很多场景中并不需要排序,在这些场景中多余的排序反而会损耗性能。
Hash Based Shuffle Write
该过程实现的核心是在 HashShuffleWriter#write(records: Iterator[Product2[K, V]]): Unit
其主要流程如下:
该函数的输入是一个 Shuffle Map Task 计算得到的结果(对应的迭代器),若在宽依赖中定义了 map 端的聚合则会先进行聚合,随后对于迭代器(若要聚合则为聚合后的迭代器)的每一项先通过计算 key 的 hash 值来确定要写到哪个文件,然后将 key、value 写入文件。
写入的文件名的格式是:shuffle_$shuffleId_$mapId_$reduceId
。写入时,若文件已存在会删除会创建新文件。
上图描述了如何处理一个 Shuffle Map Task 计算结果,在实际应用中,往往有很多 Shuffle Map Tasks 及下游 tasks,即如下情况(图摘自:JerryLead/SparkInternals-Shuffle 过程):
存在的问题
这种简单的实现会有几个问题,为说明方便,这里设 M = Shuffle Map Task 数量
,R = 下游 tasks 数量
:
- 产生过多文件:由于每个 Shuffle Map Task 需要为每个下游的 Task 创建一个单独的文件,因此文件的数量就是
M * R
。如果 Shuffle Map Tasks 数量是 1000,下游的 tasks 数是 800,那么理论上会产生 80w 个文件(对于 size 为 0的文件会特殊处理) - 打开多个文件对于系统来说意味着随机写,尤其是每个文件较小且文件特别多的情况。机械硬盘在随机读写方面的性能很差,如果是固态硬盘,会改善很多
- 缓冲区占用内存空间大:每个 Shuffle Map Task 需要开 R 个 bucket(为减少写文件次数的缓冲区),N 个 Shuffle Map Task 就会产生
N * R
个 bucket。虽然一个 Shuffle Map Task,对应的 buckets 会被回收,但一个节点上的 bucket 个数最多可以达到cores * R
个,每个 bucket 默认为 32KB。对于 24 核 1000 个 reducer 来说,占用内存就是 750MB
改进:Shuffle Consolidate Writer
在上面提到的几个问题,Spark 提供了 Shuffle Consolidate Files 机制进行优化。该机制的手段是减少 Shuffle 过程产生的文件,若使用这个功能,则需要置 spark.shuffle.consolidateFiles
为 true
,其实现可用下图来表示(图摘自:JerryLead/SparkInternals-Shuffle 过程)
即:对于运行在同一个 core 的 Shuffle Map Tasks,对于将要被同一个 reducer read 的数据,第一个 Shuffle Map Task 会创建一个文件,之后的就会将数据追加到这个文件而不是新建一个文件(相当于同一个 core 上的 Shuffle Map Task 写了文件不同的部分)。因此文件数就从原来的 M * R
个变成了 cores * R
个。当 M / cores
的值越大,减少文件数的效果越显著。需要注意的是,该机制虽然在很多时候能缓解上述的几个问题,但是并不能彻底解决。
参考
- 《Spark 技术内幕》
- JerryLead/SparkInternals - Shuffle 过程