在具体的实现上,Shuffle经历了Hash、Sort、Tungsten-Sort三阶段,shuffle共有三种,别人讨论的是hash shuffle,这是最原始的实现,曾经有两个版本:
* 第一版是每个map产生r个文件,一共产生mr个文件,由于产生的中间文件太大影响扩展性;
* 社区提出了第二个优化版本,让一个core上map共用文件,减少文件数目,这样共产生corer个文件,好多了,但中间文件数目仍随任务数线性增加,仍难以应对大作业,但hash shuffle已经优化到头了;
为了解决hash shuffle性能差的问题,又引入sort shuffle,完全借鉴mapreduce实现,每个map产生一个文件,彻底解决了扩展性问题。
网上有很多文章,发现跟当前的版本有很多冲突,看起来一知半解的;
当前讨论的版本为spark2.1.0,之前的版本不再考虑;
而其命名方式也很奇怪,可能为兼容之前的逻辑;
从Spark Shuffle的历程来看,可以分为两大类:hash和sort shuffle。
Hash ShuffleManager
Hash Shuffle在spark2.0+被移除;
SortShuffleManager
从SparkEnv中看,当前唯一支持的shuffle方式为SortShuffleManager,并且不管是sort/tungten-sort均是采用该方式。
/** Get a writer for a given partition. Called on executors by map tasks. */
override def getWriter[K, V](
handle: ShuffleHandle,
mapId: Int,
context: TaskContext): ShuffleWriter[K, V] = {
numMapsForShuffle.putIfAbsent(
handle.shuffleId, handle.asInstanceOf[BaseShuffleHandle[_, _, _]].numMaps)
val env = SparkEnv.get
handle match {
case unsafeShuffleHandle: SerializedShuffleHandle[K @unchecked, V @unchecked] =>
new UnsafeShuffleWriter(
env.blockManager,
shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver],
context.taskMemoryManager(),
unsafeShuffleHandle,
mapId,
context,
env.conf)
case bypassMergeSortHandle: BypassMergeSortShuffleHandle[K @unchecked, V @unchecked] =>
new BypassMergeSortShuffleWriter(
env.blockManager,
shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver],
bypassMergeSortHandle,
mapId,
context,
env.conf)
case other: BaseShuffleHandle[K @unchecked, V @unchecked, _] =>
new SortShuffleWriter(shuffleBlockResolver, other, mapId, context)
}
}
各writer的特点:
Writer类型 | 特点 |
---|---|
BypassMergeSortShuffleWriter | 和Hash Shuffle实现基本相同,区别在于map task输出会汇总为一个文件 |
UnsafeShuffleWriter | tungsten-sort,ShuffleExternalSorter使用Java Unsafe直接操作内存,避免Java对象多余的开销和GC 延迟,效率高 |
SortShuffleWriter S | ort Shuffle,和HashShuffle的主要不同在于,map端支持Partition级别的sort,map task输出会汇总为一个文件 |
map-side aggregation | Partition数(RDD) | Serializer支持relocation | |
---|---|---|---|
BypassMergeSortShuffleWriter | 否 | 小于200(默认) | – |
UnsafeShuffleWriter | 否 | 小于16777216 | 是 |
SortShuffleWriter | – | – | – |
各writer的使用条件:
map-side aggregation | Partition数(RDD) | Serializer支持relocation | |
---|---|---|---|
BypassMergeSortShuffleWriter | 否 | 小于200(默认) | – |
UnsafeShuffleWriter | 否 | 小于16777216 | 是 |
SortShuffleWriter | – | – | – |
其中:
- 没有map端聚合操作,且RDD的Partition数小于200,使用BypassMergeSortShuffleWriter;
- 没有map端聚合操作,RDD的Partition数小于16777216,且Serializer支持relocation,使用UnsafeShuffleWriter;
- 上述条件都不满足,使用SortShuffleWriter。
使用场景
UnsafeShuffleWriter:
- 基于ShuffleExternalSorter(内部基于ShuffleInMemorySorter实现)完成sort完成sort,最终结果汇总至一个文件;
- *ShuffleExternalSorter使用UnSafe API操作序列化数据,而不是Java对象,减少了内存占用及因此导致的GC耗时(参考Spark 内存管理之Tungsten),这个优化需要Serializer支持relocation;
- ShuffleExternalSorter存原始数据,ShuffleInMemorySorter使用压缩指针存储元数据,每条记录仅占8 bytes,并且排序时不需要处理原始数据,效率高;
- 溢写 & 合并这一步操作的是同一Partition的数据,因为使用UnSafe API直接操作序列化数据,合并时不需要反序列化数据
- 溢写 & 合并可以使用fastMerge提升效率(调用NIO的transferTo方法),设置spark.shuffle.unsafe.fastMergeEnabled为true,并且如果使用了压缩,需要压缩算法支持SerializedStreams的连接,各默认值如下
是否支持aggregation | 实现 | |
---|---|---|
PartitionedAppendOnlyMap | 支持 | 基于Array实现的HashMap结构,支持lookup,并在此基础上实现aggregation,使用线性探查法处理Hash冲突 |
PartitionedPairBuffer | 不支持 | 就是Array结构,K-V Pair依次写入数组 |
内部存储PartitionedAppendOnlyMap/PartitionedPairBuffer的区别:
是否支持aggregation | 实现 | |
---|---|---|
PartitionedAppendOnlyMap | 支持 | 基于Array实现的HashMap结构,支持lookup,并在此基础上实现aggregation,使用线性探查法处理Hash冲突 |
PartitionedPairBuffer | 不支持 | 就是Array结构,K-V Pair依次写入数组 |
BypassMergeSortShuffleWriter:
- 没有使用sorter操作(?从实现上看确实没有sort操作?),每个reduce(partition)会产生一个file,所以其有阈值限制(partition不超过200);
SortShuffleWriter:
- 基于ExternalSort,每个rdd输出一个文件(有合并过程);
参考:
- Spark Sort Based Shuffle内存分析 http://www.jianshu.com/p/c83bb237caa8
- Spark Shuffle的技术演进: http://www.jianshu.com/p/4c5c2e535da5
- Project Tungsten:让Spark将硬件性能压榨到极限: http://www.csdn.net/article/2015-04-30/2824591-project-tungsten-bringing-spark-closer-to-bare-metal
- Project Tungsten (Spark 1.5 Phase 1): https://issues.apache.org/jira/browse/SPARK-7075
- Faster sort-based shuffle path using binary processing cache-aware sort:https://issues.apache.org/jira/browse/SPARK-7081
- Spark Shuffle之Tungsten-Sort: http://blog.csdn.net/u011564172/article/details/72764010
- Spark Shuffle之Sort Shuffle: http://blog.csdn.net/u011564172/article/details/72763978
- Spark Shuffle之Hash Shuffle: http://blog.csdn.net/u011564172/article/details/71170234
- Spark Shuffle之SortShuffleWriter: http://blog.csdn.net/u011564172/article/details/72764032