HashShuffle和SortShuffleManager实现分析
在《Spark基础以及Shuffle实现分析》中，我们说到Shuffle包括ShuffleMapStage和ShuffledRDD两步骤，分别对应了Shuffle的Map和Reduce；在这两个步骤中ShuffleManager充当了很重要的角色。
- 它指导了ShuffleMapStage的Task怎么进行Map的write操作
- 它指导了ShuffledRDD的compute函数怎么去每个Map的节点拉取指定reduce的数据；
ShuffleManager接口代码这里就不贴了，详情参见上一篇文章。
另外在Spark中，通过配置信息可以来选择具体ShuffleManager的实现版本：HashShuffleManager或SortShuffleManager。下面我们分别针对两种ShuffleManager进行分析；
HashShuffleManager
HashShuffleManager是Spark最早版本的ShuffleManager，该ShuffleManager的严重缺点是会产生太多小文件，特别是reduce个数很多时候，存在很大的性能瓶颈。
这里就直接给出小文件个数，后面会分析原因：ShuffleMapTask个数×reduce个数。这个数字很大的，假设512M的HDFS的分片大小，1T数据对应2048个map，如果reduce再大， 那么就生成几万/几十万个小文件，这是致命的；
针对这个问题，HashShuffleManager在“某个版本”支持了consolidateFiles（合并文件）的机制；可以通过“spark.shuffle.consolidateFiles”配置来启动，在处理reduce个数很多的 计算时候，这个功能可以显著的提供处理性能。这个机制生成的小文件个数是多少呢？对于单个Executor，如果并发的ShuffleMapTask的个数为M个，那么在该Executor上针对该Shuffle 最多会生成M×reduce个数个小文件，总小文件的个数，当然是所有Executor上小文件之和了！这个机制可以很大程度上减少小文件的个数。
那么下面我们就来分析HashShuffleManager的具体实现。
private[spark] class HashShuffleManager(conf: SparkConf) extends ShuffleManager
private val fileShuffleBlockManager = new FileShuffleBlockManager(conf)
override def getReader[K, C](
handle: ShuffleHandle,startPartition: Int,endPartition: Int,
context: TaskContext): ShuffleReader[K, C] = {
new HashShuffleReader(handle, startPartition, endPartition, context)
}
override def getWriter[K, V](handle: ShuffleHandle, mapId: Int, context: TaskContext)
: ShuffleWriter[K, V] = {
new HashShuffleWriter(shuffleBlockManager, handle, mapId, context)
}
override def shuffleBlockManager: FileShuffleBlockManager = {
fileShuffleBlockManager
}
}
HashShuffleManager的代码很简单，把所有功能都通过FileShuffleBlockManager，HashShuffleWriter，HashShuffleReader三个类来实现，其中FileShuffleBlockManager最为重要。 它管理Shuffle运行过程中所涉及到文件，换句话说，FileShuffleBlockManager为HashShuffleWriter，HashShuffleReader提供了write和read的FileHandle。HashShuffleWriter比较简单， 但是HashShuffleReader还是比较复杂的，后面我们会分析。
FileShuffleBlockManager
上面说到consolidateFiles其实就是在FileShuffleBlockManager上进行改进，为了清晰表达FileShuffleBlockManager的功能，我们首先把consolidateFiles移除来进行讲解；
FileShuffleBlockManager它是BlockManager，它负责了Shuffle的Mapoutput的Block怎么写，以及怎么读；
针对怎么写的问题，FileShuffleBlockManager提供了forMapTask接口：
def forMapTask(shuffleId: Int, mapId: Int, numBuckets: Int, serializer: Serializer,
writeMetrics: ShuffleWriteMetrics) = {
new ShuffleWriterGroup {
val writers: Array[BlockObjectWriter] ={
Array.tabulate[BlockObjectWriter](numBuckets) { bucketId =>
val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
val blockFile = blockManager.diskBlockManager.getFile(blockId)
if (blockFile.exists) {
if (blockFile.delete()) {
logInfo(s"Removed existing shuffle file $blockFile")
} else {
logWarning(s"Failed to remove existing shuffle file $blockFile")
}
}
blockManager.getDiskWriter(blockId, blockFile, serializer, bufferSize, writeMetrics)
}
}
}
}
这个函数看起来很简单，它针对每个MapTask构造一个ShuffleWriterGroup对象，该对象包含一组writer，个数为numBuckets，即reduce个数，这里对文件的管理基于 blockManager.diskBlockManager来实现的，详细可以参见《 Spark-Block管理》的讲解；
这里我们可以看到，对应每个Map都创建numBuckets个小文件，从而证明了“ShuffleMapTask个数×reduce个数”
有写就有读，getBlockData接口提供了读取指定ShuffleBlockID的数据
override def getBlockData(blockId: ShuffleBlockId): ManagedBuffer = {
val file = blockManager.diskBlockManager.getFile(blockId)
new FileSegmentManagedBuffer(file, 0, file.length)
}
从上面我们看到，不支持consolidateFiles的FileShuffleBlockManager还是很简单。下面我们分析支持consolidateFiles的FileShuffleBlockManager的实现
支持consolidateFiles的FileShuffleBlockManager
从上面forMapTask我们看到，针对每个Map都会创建一个ShuffleWriterGroup对象，其中封装了一组文件的writer，这个时候我们想，多个mapTask是否可以向这个 ShuffleWriterGroup对象进行写？是的，可以，但是有一个前提是前一个map必须写完了，即close了该ShuffleWriterGroup所有的writer，那么下一个map就可以直接使用 这组ShuffleWriterGroup对象所对应Files进行写；然后通过offset和length来从相同的文件中读取属于自己map的这部分。
上面说的有点啰嗦，下面我们一步一步进行解析。
对consolidateFiles的理解首先必须理解FileGroup的概念，代码如下：
private class ShuffleFileGroup(val shuffleId: Int, val fileId: Int, val files: Array[File]) {
private var numBlocks: Int = 0
private val mapIdToIndex = new PrimitiveKeyOpenHashMap[Int, Int]()
private val blockOffsetsByReducer = Array.fill[PrimitiveVector[Long]](files.length) {
new PrimitiveVector[Long]()
}
private val blockLengthsByReducer = Array.fill[PrimitiveVector[Long]](files.length) {
new PrimitiveVector[Long]()
}
def apply(bucketId: Int) = files(bucketId)
def recordMapOutput(mapId: Int, offsets: Array[Long], lengths: Array[Long]) {
mapIdToIndex(mapId) = numBlocks
numBlocks += 1
for (i <- 0 until offsets.length) {
blockOffsetsByReducer(i) += offsets(i)
blockLengthsByReducer(i) += lengths(i)
}
}
def getFileSegmentFor(mapId: Int, reducerId: Int): Option[FileSegment] = {
val file = files(reducerId)
val blockOffsets = blockOffsetsByReducer(reducerId)
val blockLengths = blockLengthsByReducer(reducerId)
val index = mapIdToIndex.getOrElse(mapId, -1)
if (index >= 0) {
val offset = blockOffsets(index)
val length = blockLengths(index)
Some(new FileSegment(file, offset, length))
} else {
None
}
}
}
每个ShuffleFileGroup都被赋予了唯一的编号，即fileId，它封装了一组文件的Handle，即val files: Array[File]，它的个数即reduce的个数。
ShuffleFileGroup中最为重要四个属性:
- numBlocks:当前ShuffleFileGroup被写入了多少个Block，这里多少个什么含义？如果有10个MapTask使用了这组ShuffleFileGroup，那么numBlocks就为10；所以 这里numBlocks命名是不好的；因为按照BlockManager里面对Block的定义，每个ShuffleBlock应该是【"shuffle" + shuffleId + "" + mapId + "_" + reduceId】而这里把Map 当着Block单位；
- mapIdToIndex：这里根据使用该ShuffleFileGroup的mapTask先后顺序进行编号，比如mapID=4，mapID=7先后使用了该FileGroup，那么4->0,7->1的映射就存储在mapIdToIndex 中；
- blockOffsetsByReducer/blockLengthsByReducer：现在有多个MapTask使用追加的方式使用同一组文件，对于其中一个reduce文件，那么每个Map就使用该文件的offset+length区间段的 文件，因此，对于每个reduce文件，我们需要存储每个mapTask使用该文件的offset和length，即这里的blockOffsetsByReducer/blockLengthsByReducer； 这两个可以理解为一个二维数组，一维为reduce编号，所以一维的个数是确定的，即reduce的个数；每个二维是一个vector，它的个数等于numBlocks，即使用该FileGroup的MapTask的个数；
对上面四个属性的理解以后，对recordMapOutput和getFileSegmentFor这两个接口理解比较简单了，分别对于了更新操作和读取操作，就不详细说了，很简单；
对ShuffleFileGroup的理解是理解consolidateFiles的重点；在传统的FileShuffleBlockManager中，每个MapTask私有一组File，即一个FileGroup，而在consolidateFiles中， 多个Map可以按照先后顺序共有一组File；
下面我们就来分析，在Spark中，是怎么实现这个先后顺序来共有一个ShuffleFileGroup；
private class ShuffleState(val numBuckets: Int) {
val unusedFileGroups = new ConcurrentLinkedQueue[ShuffleFileGroup]()
val allFileGroups = new ConcurrentLinkedQueue[ShuffleFileGroup]()
}
对于每个Shuffle，在FileShuffleBlockManager维护一个ShuffleState的对象，该对象有两个ShuffleFileGroup的队列（注意是队列，为什么？）;
- allFileGroups:存储该Shuffle使用到所有的ShuffleFileGroup,所以对于Executor上,每个Shuffle使用的文件总数就为allFileGroups.length * reduce个数;
- unusedFileGroups:一个ShuffleFileGroup在被一个MapTask使用完(使用完就是MapTask运行结束,关闭了writer)以后,会将它放到unusedFileGroups中,从而可以被重用. 所以这里的unused不是废弃的含义;另外上面说了这个是队列,而不是列表和set之类的,先进先出从而可以平均每个ShuffleFileGroup被重用的次数,否则会出现一个ShuffleFileGroup被重复使用N次;
到目前为止我们差不多已经知道了支持consolidateFiles的FileShuffleBlockManager是怎么实现的,我们还是来看一下代码的具体的实现;
def forMapTask(shuffleId: Int, mapId: Int, numBuckets: Int, serializer: Serializer,
writeMetrics: ShuffleWriteMetrics) = {
new ShuffleWriterGroup {
shuffleStates.putIfAbsent(shuffleId, new ShuffleState(numBuckets))
private val shuffleState = shuffleStates(shuffleId)
private var fileGroup: ShuffleFileGroup = null
val writers: Array[BlockObjectWriter] = {
fileGroup = getUnusedFileGroup()
Array.tabulate[BlockObjectWriter](numBuckets) { bucketId =>
val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
blockManager.getDiskWriter(blockId, fileGroup(bucketId), serializer, bufferSize,
writeMetrics)
}
}
override def releaseWriters(success: Boolean) {
if (success) {
val offsets = writers.map(_.fileSegment().offset)
val lengths = writers.map(_.fileSegment().length)
fileGroup.recordMapOutput(mapId, offsets, lengths)
}
recycleFileGroup(fileGroup)
}
private def getUnusedFileGroup(): ShuffleFileGroup = {
val fileGroup = shuffleState.unusedFileGroups.poll()
if (fileGroup != null) fileGroup else newFileGroup()
}
private def newFileGroup(): ShuffleFileGroup = {
val fileId = shuffleState.nextFileId.getAndIncrement()
val files = Array.tabulate[File](numBuckets) { bucketId =>
val filename = physicalFileName(shuffleId, bucketId, fileId)
blockManager.diskBlockManager.getFile(filename)
}
val fileGroup = new ShuffleFileGroup(shuffleId, fileId, files)
shuffleState.allFileGroups.add(fileGroup)
fileGroup
}
private def recycleFileGroup(group: ShuffleFileGroup) {
shuffleState.unusedFileGroups.add(group)
}
}
}
这是支持consolidateFiles的forMapTask的实现,和上面不同,在创建writers的时候,通过调用getUnusedFileGroup来获得; 在getUnusedFileGroup首先尝试从队列中获取 一个unusedFileGroup,如果失败就newFileGroup一个,从而实现重用; 那么一个fileGroup怎么进入unusedFileGroup呢?即releaseWriters,如果当前MapTask运行成功, 会把该MapTask对ShuffleFile的写的offset以及length更新到fileGroup中,同时设置为unusedFileGroup;
override def getBlockData(blockId: ShuffleBlockId): ManagedBuffer = {
// Search all file groups associated with this shuffle.
val shuffleState = shuffleStates(blockId.shuffleId)
val iter = shuffleState.allFileGroups.iterator
while (iter.hasNext) {
val segmentOpt = iter.next.getFileSegmentFor(blockId.mapId, blockId.reduceId)
if (segmentOpt.isDefined) {
val segment = segmentOpt.get
return new FileSegmentManagedBuffer(segment.file, segment.offset, segment.length)
}
}
throw new IllegalStateException("Failed to find shuffle block: " + blockId)
}
getBlockData接口就需要去遍历所有的FileGroup, 即allFileGroups,判读其中是否有指定的mapID和reduceID对应的数据,如果有就结束遍历;
OK,我想我们应该理解了支持consolidateFiles的FileShuffleBlockManager的实现了;
HashShuffleWriter的实现;
如果对FileShuffleBlockManager的理解比较清楚,HashShuffleWriter的理解就比较简单.
private[spark] class HashShuffleWriter[K, V](
shuffleBlockManager: FileShuffleBlockManager,
handle: BaseShuffleHandle[K, V, _],
mapId: Int,
context: TaskContext)
extends ShuffleWriter[K, V] with Logging {
private val dep = handle.dependency
private val numOutputSplits = dep.partitioner.numPartitions
private val shuffle = shuffleBlockManager.forMapTask(dep.shuffleId, mapId, numOutputSplits, ser,
writeMetrics)
}
每个Map都对应一个HashShuffleWriter, 通过reduce的分区函数partitioner来确定reduce的个数, 然后通过上面的forMapTask来返回一组FileHandle.
另外我们知道MapReduce中有一个Map端的Combine机制, 即mapSideCombine, 如果需要,那么就需要在write之前进行reduce操作, 详细逻辑如下, dep.aggregator.get.combineValuesByKey就为mapSideCombine的逻辑;
override def write(records: Iterator[_ <: Product2[K, V]]): Unit = {
val iter = if (dep.aggregator.isDefined) {
if (dep.mapSideCombine) {
dep.aggregator.get.combineValuesByKey(records, context)
} else {
records
}
} else if (dep.aggregator.isEmpty && dep.mapSideCombine) {
throw new IllegalStateException("Aggregator is empty for map-side combine")
} else {
records
}
for (elem <- iter) {
val bucketId = dep.partitioner.getPartition(elem._1)
shuffle.writers(bucketId).write(elem)
}
}
我们知道每个MapTask运行结束以后,需要请求FileShuffleBlockManager关闭相应的writers,并向Driver返回一个MapStatus,从而可以被reduce操作定位Map的位置, 这部分逻辑就是commitWritesAndBuildStatus来实现的
private def commitWritesAndBuildStatus(): MapStatus = {
val sizes: Array[Long] = shuffle.writers.map { writer: BlockObjectWriter =>
writer.commitAndClose()
writer.fileSegment().length
}
MapStatus(blockManager.blockManagerId, sizes)
}
shuffle.releaseWriters(success)
在commitWritesAndBuildStatus中统计每个reduce的大小,并关闭每个reduce的writer, 最后返回MapStatus; 在commitWritesAndBuildStatus执行完成以后就请求FileShuffleBlockManager调用releaseWriter,关闭FileGroup;
所以整体来说HashShuffleWriter的实现还是很简单的; 通过这里的分析,大家应该对《Spark基础以及Shuffle实现分析》中ShuffleMapStage应该有更 深入的认识了;
HashShuffleReader的实现;
这里我不打算继续讲HashShuffleReader,为什么?如果大家打开SortShuffleManager,我们就会发现, 它也是使用HashShuffleReader. 这么说就是HashShuffleReader 和具体的ShuffleManager无关,在分析完SortShuffleManager我们再统一进行分析;
SortShuffleManager
在HashShuffleManager中,不管是否支持consolidateFiles, 同一个map的多个reduce之间都对应了不同的文件,至于对应哪个文件,是由分区函数进行Hash来确定的; 这是为什么它要叫做HashShuffleManager.
下面要介绍的SortShuffleManager和HashShuffleManager有一个本质的差别,即同一个map的多个reduce的数据都写入到同一个文件中;那么SortShuffleManager产生的Shuffle 文件个数为2*Map个数; 不是说,每个map只对应一个文件吗?为什么要乘以2呢?下面我们来一一分析;
private[spark] class SortShuffleManager(conf: SparkConf) extends ShuffleManager {
private val indexShuffleBlockManager = new IndexShuffleBlockManager()
private val shuffleMapNumber = new ConcurrentHashMap[Int, Int]()
override def getReader[K, C](
handle: ShuffleHandle,startPartition: Int,endPartition: Int,context: TaskContext): ShuffleReader= {
new HashShuffleReader(
handle.asInstanceOf[BaseShuffleHandle[K, _, C]], startPartition, endPartition, context)
}
override def getWriter[K, V](handle: ShuffleHandle, mapId: Int, context: TaskContext)
: ShuffleWriter[K, V] = {
val baseShuffleHandle = handle.asInstanceOf[BaseShuffleHandle[K, V, _]]
shuffleMapNumber.putIfAbsent(baseShuffleHandle.shuffleId, baseShuffleHandle.numMaps)
new SortShuffleWriter(
shuffleBlockManager, baseShuffleHandle, mapId, context)
}
override def shuffleBlockManager: IndexShuffleBlockManager = {
indexShuffleBlockManager
}
}
和HashShuffleManager一下, SortShuffleManager的主要的功能是依赖IndexShuffleBlockManager, HashShuffleReader, SortShuffleWriter三个类来实现;其中IndexShuffleBlockManager 管理Shuffle的文件,而和HashShuffleManager不一样,IndexShuffleBlockManager的复杂度很小,反而SortShuffleWriter是SortShuffleManager的核心, 它实现了一个map的所有的reduce数据怎么写到一个文件中(先透露一下,对该map的数据进行外部排序,这样属于每个reduce的数据就为排序后文件的一个文件段);而HashShuffleReader 的实现和HashShuffleManager一致;
下面我们先分析IndexShuffleBlockManager;
@DeveloperApi
case class ShuffleDataBlockId(shuffleId: Int, mapId: Int, reduceId: Int) extends BlockId {
def name = "shuffle_" + shuffleId + "_" + mapId + "_" + reduceId + ".data"
}
@DeveloperApi
case class ShuffleIndexBlockId(shuffleId: Int, mapId: Int, reduceId: Int) extends BlockId {
def name = "shuffle_" + shuffleId + "_" + mapId + "_" + reduceId + ".index"
}
首先,在IndexShuffleBlockManager中,针对每个Map有两个文件,一个是index文件,一个是data文件;这个时候你肯定会问,这里的ShuffleDataBlockId,ShuffleIndexBlockId 所对应的blockID, 即为最后文件的name, 其中不是把reduceID也作为文件名的一部分吗?难道我说错了?每个map的每个reduce都对应一个Index文件和一个Data文件?
当然不是这样的,那么下面我们就来看IndexShuffleBlockManager的实现
private[spark]
class IndexShuffleBlockManager extends ShuffleBlockManager {
private lazy val blockManager = SparkEnv.get.blockManager
def getDataFile(shuffleId: Int, mapId: Int): File = {
blockManager.diskBlockManager.getFile(ShuffleDataBlockId(shuffleId, mapId, 0))
}
private def getIndexFile(shuffleId: Int, mapId: Int): File = {
blockManager.diskBlockManager.getFile(ShuffleIndexBlockId(shuffleId, mapId, 0))
}
def removeDataByMap(shuffleId: Int, mapId: Int): Unit = {
getDataFile(shuffleId, mapId).delete()
getIndexFile(shuffleId, mapId).delete()
}
def writeIndexFile(shuffleId: Int, mapId: Int, lengths: Array[Long]) = {
val indexFile = getIndexFile(shuffleId, mapId)
val out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(indexFile)))
var offset = 0L
out.writeLong(offset)
for (length <- lengths) {
offset += length
out.writeLong(offset)
}
}
override def getBlockData(blockId: ShuffleBlockId): ManagedBuffer = {
val indexFile = getIndexFile(blockId.shuffleId, blockId.mapId)
val in = new DataInputStream(new FileInputStream(indexFile))
in.skip(blockId.reduceId * 8)
val offset = in.readLong()
val nextOffset = in.readLong()
new FileSegmentManagedBuffer(getDataFile(blockId.shuffleId, blockId.mapId),
offset,nextOffset - offset)
}
}
首先从getDataFile和getIndexFile文件我们看到,获取ShuffleBlockID的时候是把reduceID设置为0, 换句话这就证明上面那个疑问是错误的;然后我就看getBlockData, 该函数的功能是从当前的ShuffleBlockManager中读取到指定Shuffle reduce的内容:从逻辑上面我们看到, 首先从indexFile中,按照blockId.reduceId * 8的offset 开始读取一个Long数据,这个Long就代表当前reduce数据在getDataFile中偏移量;再看writeIndexFile,我们就看到,针对一个Map都会生成一个Index文件, 按照reduce的顺序将她们的offset输出到index文件中;
到这里你肯定Data文件是怎么输出的?难道是删除了Data文件写的逻辑?不是的;IndexShuffleBlockManager的确只提供了对Index文件管理,至于Data的文件怎么写?那就我们 接下来需要分析的SortShuffleWriter的实现了;
private[spark] class SortShuffleWriter[K, V, C](
shuffleBlockManager: IndexShuffleBlockManager,
handle: BaseShuffleHandle[K, V, C], mapId: Int, context: TaskContext)
extends ShuffleWriter[K, V] with Logging {
private val dep = handle.dependency
private val blockManager = SparkEnv.get.blockManager
private var sorter: ExternalSorter[K, V, _] = null
override def write(records: Iterator[_ <: Product2[K, V]]): Unit = {
if (dep.mapSideCombine) {
sorter = new ExternalSorter[K, V, C](
dep.aggregator, Some(dep.partitioner), dep.keyOrdering, dep.serializer)
sorter.insertAll(records)
} else {
sorter = new ExternalSorter[K, V, V](
None, Some(dep.partitioner), None, dep.serializer)
sorter.insertAll(records)
}
val outputFile = shuffleBlockManager.getDataFile(dep.shuffleId, mapId)
val blockId = shuffleBlockManager.consolidateId(dep.shuffleId, mapId)
val partitionLengths = sorter.writePartitionedFile(blockId, context, outputFile)
shuffleBlockManager.writeIndexFile(dep.shuffleId, mapId, partitionLengths)
mapStatus = MapStatus(blockManager.blockManagerId, partitionLengths)
}
}
SortShuffleWriter的实现依赖ExternalSorter.从名称我们就看到它的含义是外部排序的含义, 我们调用sorter.writePartitionedFile(blockId, context, outputFile)将当前分片的 reduce数据输出到getDataFile(dep.shuffleId, mapId)文件中, 并返回每个reduce的大小, 然后调用writeIndexFile来写index文件;
从这里我们看到, SortShuffleManager本质上对我们的数据进行外部排序,并把排序后文件分为每个reduce段, 并把每个端的偏移量写到index文件中;至于外部排序的具体实现 我就不去分析了.
现在还ShuffleManager里面还是最后一个大问题就是ShuffleRead,下面我们继续分析;
ShuffleReader
在《Spark基础以及Shuffle实现分析》中,我对ShuffledRDD的实现也是简单进行解释了一下,这里通过对ShuffleReader的实现的分析,对Shuffle 的Reduce步骤进行详细分析;
class ShuffledRDD[K, V, C](
@transient var prev: RDD[_ <: Product2[K, V]],
part: Partitioner)
extends RDD[(K, C)](prev.context, Nil) {
override def compute(split: Partition, context: TaskContext): Iterator[(K, C)] = {
val dep = dependencies.head.asInstanceOf[ShuffleDependency[K, V, C]]
SparkEnv.get.shuffleManager.getReader(dep.shuffleHandle, split.index, split.index + 1, context)
.read()
.asInstanceOf[Iterator[(K, C)]]
}
}
上面ShuffledRDD的compute函数,从逻辑我们可以看到, 它是通过shuffleManager.getReader的函数来实现了, 即我们这里要分析的ShuffleReader;
我们知道ShuffleMapStage中所有ShuffleMapTask是分散在Executor上的, 每个Map对应一个Task, Task运行结束以后, 会把MapOutput的信息保存在MapStatus返回给Driver, Driver将其注册到MapOutputTrack中; 到目前为止, ShuffleMapStage的过程就执行完成了;
ShuffledRDD会为每个reduce创建一个分片, 对于运行在Executor A上的shuffleRDD的一个分片的Task, 为了获取该分片的对于的Reduce数据, 它需要向MapOutputTrack获取 指定ShuffleID的所有的MapStatus, 由于MapOutputTrack是一个主从结构, 获取MapStatus也涉及到Executor A请求Driver的过程. 一旦获得该Shuffle所对于的所有的MapStatus, 该Task从每个MapStatus所对应的Map节点(BlockManager节点)去拉取指定reduce的数据, 并把所有的数据组合为Iterator, 从而完成ShuffledRDD的compute的过程;
上面即ShuffledRDD的compute过程的描述,下面我们来一步一步看每步的实现, 同时找出原因,为什么针对不同的ShuffleManager,都可以使用同一个ShuffleReader: HashShuffleReader
private[spark] class HashShuffleReader[K, C](
handle: BaseShuffleHandle[K, _, C],startPartition: Int,endPartition: Int,
context: TaskContext)extends ShuffleReader[K, C] {
private val dep = handle.dependency
override def read(): Iterator[Product2[K, C]] = {
val ser = Serializer.getSerializer(dep.serializer)
val iter = BlockStoreShuffleFetcher.fetch(handle.shuffleId, startPartition, context, ser)
val aggregatedIter: Iterator[Product2[K, C]] = if (dep.aggregator.isDefined) {
if (dep.mapSideCombine) {
new InterruptibleIterator(context, dep.aggregator.get.combineCombinersByKey(iter, context))
} else {
new InterruptibleIterator(context, dep.aggregator.get.combineValuesByKey(iter, context))
}
} else if (dep.aggregator.isEmpty && dep.mapSideCombine) {
throw new IllegalStateException("Aggregator is empty for map-side combine")
} else {
iter.asInstanceOf[Iterator[Product2[K, C]]].map(pair => (pair._1, pair._2))
}
aggregatedIter
}
从HashShuffleReader的代码来看, 它的逻辑很简洁, 其中核心就是通过BlockStoreShuffleFetcher.fetch获取指定reduce的所有的数据, 因此fetch就是我们上面说去每个 Map拉取数据的过程.
private[hash] object BlockStoreShuffleFetcher extends Logging {
def fetch[T](shuffleId: Int,reduceId: Int,context: TaskContext,serializer: Serializer)
: Iterator[T] ={
val blockManager = SparkEnv.get.blockManager
val statuses = SparkEnv.get.mapOutputTracker.getServerStatuses(shuffleId, reduceId)
val blockFetcherItr = new ShuffleBlockFetcherIterator(
context, SparkEnv.get.blockTransferService,blockManager,blocksByAddress,
serializer,SparkEnv.get.conf.getLong("spark.reducer.maxMbInFlight", 48) * 1024 * 1024)
val itr = blockFetcherItr.flatMap(unpackBlock)
new InterruptibleIterator[T](context, itr)
}
}
BlockStoreShuffleFetcher里面也没有完成实现fetch逻辑,而是交给ShuffleBlockFetcherIterator进行fetch, 但是获取指定Shuffle的所有reduce的MapStatus是这步进行的 即调用SparkEnv.get.mapOutputTracker.getServerStatuses(shuffleId, reduceId)来实现的;
那么下面就剩下了ShufflerRead最后一个步骤了,即 ShuffleBlockFetcherIterator的实现;
ShuffleBlockFetcherIterator接受的参数中一个参数就是指定的Shuffle的所有MapStatus信息,每个MapStatus其实是代表一个BlockManager的地址信息;那么就存在Local和Remote的 差别,如果是Local就可以直接通过本地的ShuffleBlockManager读取, 否则就去发起请求去远程读取.
因此第一个问题就是怎么对MapStatus进行划分, 划分为Local和Remote两个类别,即ShuffleBlockFetcherIterator. splitLocalRemoteBlocks的实现
private[this] def splitLocalRemoteBlocks(): ArrayBuffer[FetchRequest] = {
val remoteRequests = new ArrayBuffer[FetchRequest]
var totalBlocks = 0
for ((address, blockInfos) <- blocksByAddress) {
totalBlocks += blockInfos.size
if (address == blockManager.blockManagerId) {
localBlocks ++= blockInfos.filter(_._2 != 0).map(_._1)
numBlocksToFetch += localBlocks.size
} else {
val iterator = blockInfos.iterator
var curRequestSize = 0L
var curBlocks = new ArrayBuffer[(BlockId, Long)]
while (iterator.hasNext) {
val (blockId, size) = iterator.next()
if (size > 0) {
curBlocks += ((blockId, size))
remoteBlocks += blockId
numBlocksToFetch += 1
curRequestSize += size
}
if (curRequestSize >= targetRequestSize) {
remoteRequests += new FetchRequest(address, curBlocks)
curBlocks = new ArrayBuffer[(BlockId, Long)]
curRequestSize = 0
}
}
// Add in the final request
if (curBlocks.nonEmpty) {
remoteRequests += new FetchRequest(address, curBlocks)
}
}
}
remoteRequests
}
splitLocalRemoteBlocks有两个过程, 第一判读当前MapStatus是不是Local,是就添加到localBlocks中, 否则就针对remote构造FetchRequest消息包并返回.
那么我们下面再一个一个来分析, Local的ShuffleBlock是怎么读取的?
private[this] def fetchLocalBlocks() {
for (id <- localBlocks) {
shuffleMetrics.localBlocksFetched += 1
results.put(new FetchResult(
id, 0, () => blockManager.getLocalShuffleFromDisk(id, serializer).get))
}
}
def getLocalShuffleFromDisk(blockId: BlockId, serializer: Serializer): Option[Iterator[Any]] = {
val buf = shuffleManager.shuffleBlockManager.getBlockData(blockId.asInstanceOf[ShuffleBlockId])
val is = wrapForCompression(blockId, buf.inputStream())
Some(serializer.newInstance().deserializeStream(is).asIterator)
}
我们看到LocalBlock的获取最后就是基于每个ShuffleManager的ShuffleBlockManager的getBlockData的来实现, 在上面ShuffleManager中,我们就分析了具体的getBlockData的实现.
那remoteBlock是怎么获取的?
private[this] def sendRequest(req: FetchRequest) {
bytesInFlight += req.size
val sizeMap = req.blocks.map { case (blockId, size) => (blockId.toString, size) }.toMap
val blockIds = req.blocks.map(_._1.toString)
blockTransferService.fetchBlocks(req.address.host, req.address.port, blockIds,
new BlockFetchingListener {
override def onBlockFetchSuccess(blockId: String, data: ManagedBuffer): Unit = {
results.put(new FetchResult(BlockId(blockId), sizeMap(blockId),
() => serializer.newInstance().deserializeStream(
blockManager.wrapForCompression(BlockId(blockId), data.inputStream())).asIterator
))
shuffleMetrics.remoteBytesRead += data.size
shuffleMetrics.remoteBlocksFetched += 1
}
override def onBlockFetchFailure(e: Throwable): Unit = {
for ((blockId, size) <- req.blocks) {
results.put(new FetchResult(blockId, -1, null))
}
}
}
)
}
我们看到远程获取Block其实是基于blockTransferService来实现的,其实这个内容是BlockManager的BlockTransferService,这里我就不具体去分析,下一步我会写一个关于BlockTransferService 的实现,其实这块在0.91就遇到一个一个Bug,就是fetch永远等待导致job假死;后面再具体分析
到目前我们应该了解了ShuffleReader的功能, 级别上两种ShuffleManager的实现,我们就分析完了, 从上面分析我们可以看到SortShuffleManager创建小文件的数目应该是 最小的,而且Map输出是有序的, 在reduce过程中如果要进行有序合并, 代价也是最小的. 也因此SortShuffleManager现在是Spark1.1版本以后的默认配置项;
另外我们经常在各种交流说到,做了什么Shuffle优化, 从上面我们看到, Shuffle针对特定的应用有很大的优化的空间,比如基于内存的Shuffle等;懂了ShuffleManager,以后听 那些交流以后, 也就不会觉得人家多么高大不可攀了,嘿嘿,因为具体的优化接口Spark都是提供的,只是针对特定应用做了自己的实现而已;