[Spark源码浅析]-关于sort-based shuffle

在sort-based shuffle,记录根据目标分区id做排序,然后写到一份单一的输出文件中。 Reducers拉取这个文件连续的region 来读取这个输出的位置。

如果这次map的输出数据 太大到不能放在内存中。这个排序输出 会被切分到磁盘。然后被合并到一个最终的文件中。

sort-based shuffle 有两个不同的输出路径 来存在输出文件。

一、序列化的排序（满足以下三个条件时）

1.shuffle依赖没有聚合或者排序输出。

2.shuffle序列化序列化值再布置。（这个已经被KryoSerializer和spark sql 传统序列化支持）

3.shuffle输出产生少于16777216个输出分区。

二、非序列化排序。处理所有类型。

相关类：

ShuffleWriter（抽象类）

写一序列化记录到这个任务的输出。
def write(records: Iterator[Product2[K, V]]): Unit

关闭这个输出。这个map输出完成会跳过。
def stop(success: Boolean): Option[MapStatus]

ShuffleWriter 有 三个子类

BypassMergeSortShuffleWriter

SortShuffleWriter

UnsafeShuffleWriter

看这个类：

private[spark] class SortShuffleWriter[K, V, C](
shuffleBlockResolver: IndexShuffleBlockResolver,
handle: BaseShuffleHandle[K, V, C],
mapId: Int,
context: TaskContext)
extends ShuffleWriter[K, V] with Logging

IndexShuffleBlockResolver

创建和维持这个shuffle block的 逻辑block 和 物理文件地址的mapping。

来自于相同的map任务的shuffle 的块的数据 被存在在一个单一的合并的数据文件中。

这个数据块的偏移地址呗存储在一个分开的索引文件中[ShuffleIndexRecord]。

我们用一个shuffle数据的shuffleBlockId 和 reduce ID 设置为0 然后加上.data作为这个数据文件的文件名。索引文件用.index作为文件名的后缀。

ExternalShuffleBlockResolver

在common的network-shuffle包里。org.apache.spark.network.shuffle

ShuffleBlockFetcherIterator

一个拉取多个block的迭代器。拉取本地block时，它从本地block manager拉取。当拉取远程block时，用提供的BlockTransferService 拉取数据。

这个创建一个tuples（blockID，InputStream）的迭代器。使得呼叫着可以处理一个管道式的接受数据。

这个实现远程拉取的节流。所以他们不会超过最大的字节限制防止使用过多的内存。