概述

一个Spark的Job分为多个stage，最后一个stage会包括一个或多个ResultTask，前面的stages会包括一个或多个ShuffleMapTasks。

ResultTask执行并将结果返回给driver application。

ShuffleMapTask将task的output根据task的partition分离到多个buckets里。一个ShuffleMapTask对应一个ShuffleDependency的partition，而总partition数同并行度、reduce数目是一致的。

Task

Task的代码在scheduler package下。

抽象类Task构造参数如下：

[java] view
plaincopy

1. private[spark] abstract class Task[T](val stageId: Int, var partitionId: Int) extends Serializable  

Task对应一个stageId和partitionId。

提供runTask()接口、kill()接口等。

提供killed变量、TaskMetrics变量、TaskContext变量等。

除了上述基本接口和变量，Task的伴生对象提供了序列化和反序列化应用依赖的jar包的方法。原因是Task需要保证工作节点具备本次Task需要的其他依赖，注册到SparkContext下，所以提供了把依赖转成流写入写出的方法。

Task的两种实现

ShuffleMapTask

ShuffleMapTask构造参数如下，

[java] view
plaincopy

1. private[spark] class ShuffleMapTask(  
2.     stageId: Int,  
3.     var rdd: RDD[_],  
4.     var dep: ShuffleDependency[_,_],  
5.     _partitionId: Int,  
6.     @transient private var locs: Seq[TaskLocation])  
7.   extends Task[MapStatus](stageId, _partitionId)  

RDD partitioner对应的是ShuffleDependency。

ShuffleMapTask复写了MapStatus向外读写的方法，因为向外读写的内容包括：stageId，rdd，dep，partitionId，epoch和split(某个partition)。对于其中的stageId，rdd，dep有统一的序列化和反序列化操作并会cache在内存里，再放到ObjectOutput里写出去。序列化操作使用的是Gzip，序列化信息会维护在serializedInfoCache = newHashMap[Int,
Array[Byte]]。这部分需要序列化并保存的原因是：stageId，rdd，dep真正代表了本次Shuffle Task的信息，为了减轻master节点负担，把这部分序列化结果cache了起来。

Stage执行逻辑

主要步骤如下：

[java] view
plaincopy

1. val ser = Serializer.getSerializer(dep.serializer)  
2. shuffle = shuffleBlockManager.forMapTask(dep.shuffleId, partitionId, numOutputSplits, ser)  

这一步是初始化一个ShuffleWriterGroup，Group里面是一个BlockObjectWriter数组。

[java] view
plaincopy

1. for (elem <- rdd.iterator(split, context)) {  
2. val pair = elem.asInstanceOf[Product2[Any, Any]]  
3.   val bucketId = dep.partitioner.getPartition(pair._1)  
4.   shuffle.writers(bucketId).write(pair)  
5. }  

这一步是为每个Writer对应一个bucket，调用每个BlockObjectWriter的write()方法写数据

[java] view
plaincopy

1. var totalBytes = 0L  
2. var totalTime = 0L  
3. val compressedSizes: Array[Byte] =   
4. shuffle.writers.map { writer: BlockObjectWriter =>  
5.     writer.commit()  
6.     writer.close()  
7. val size = writer.fileSegment().length  
8.     totalBytes += size  
9. totalTime += writer.timeWriting()  
10. MapOutputTracker.compressSize(size)  
11. }  

这一步是执行writer.commit()，并得到结果file segment大小，对总大小压缩

[java] view
plaincopy

1. val shuffleMetrics = new ShuffleWriteMetrics  
2. shuffleMetrics.shuffleBytesWritten = totalBytes  
3. shuffleMetrics.shuffleWriteTime = totalTime  
4. metrics.get.shuffleWriteMetrics = Some(shuffleMetrics)  
5.   
6. success = true  
7. new MapStatus(blockManager.blockManagerId, compressedSizes)  

这一步是记录metrcis信息，最后返回一个MapStatus类，里面是本地ShuffleMapTask结果的相关信息。

最后会release writers，让对应的shuffle文件得到记录和重用(ShuffleBlockManager管理这些file，这些file是Shuffle Task中一组Writer写的对象)。

主要把下图看懂。

重要类

ShuffleBlockManager

整体梳理：

ShuffleState维护了两个ShuffleFileGroup的ConcurrentLinkedQueue，以记录目前shuffle的state。

ShuffleState记录了一次shuffle操作的文件组状态，在ShuffleBlockManager内用Map为每个shuffleId维护了一个ShuffleState。

每个shuffleId通过forMapTask()方法得到一组writer，即ShuflleWriterGroup。这组里的writers共享一个shuffleId和mapId，但是每个对应不同的bucketId和file。在为writer分配FileGroup的时候，会从shuffleId对应的shuffle state里先取unusedFileGroup，如果不存在，则在HDFS上新建File。

对于HDFS上的目标file，writer是可以append写的。在新建file的时候，是根据shuffleId和bucket number和一个递增的fileId来创建新的文件的。

ShuffleFileGroup的重用files和记录mapId，index，offset这块似懂非懂。

重要方法：

[java] view
plaincopy

1. def forMapTask(shuffleId: Int, mapId: Int, numBuckets: Int, serializer: Serializer) = { new ShuffleWriterGroup {} }  

该方法被一个ShuffleMapTask调用，传入了这次shuffle操作的id，mapId是partitionId。Buckects数目等于分区数目。该方法返回的ShuffleWriterGroup里面是一组DiskBlockObjectWriter，每一个writer都属于这一次shuffle操作，所以他们有共同的shuffleId，mapId，但是他们对应了不同的bucket，并且各自对应一个file。

在shuffle run里的调用和参数传入：

[java] view
plaincopy

1. val ser = Serializer.getSerializer(dep.serializer)  
2. shuffle = shuffleBlockManager.forMapTask(dep.shuffleId, partitionId, numOutputSplits, ser)  

shuffleId是由ShuffleDependency获得的全局唯一id，代表本次shuffle任务id

mapId等于partitionId

Bucket数目等于分区数目

产生writers：

Writer类型是DiskBlockObjectWriter，数目等于buckets数目。bufferSize的设置：

[java] view
plaincopy

1. conf.getInt("spark.shuffle.file.buffer.kb", 100) * 1024  

blockId产生自：

[java] view
plaincopy

1. blockId = ShuffleBlockId(shuffleId, mapId, bucketId)  

在生成writer的时候调用的是BlockManager的getDiskWriter方法，ShuffleBlockManager初始化的时候绑定BlockManager。

[java] view
plaincopy

1. private[spark] class DiskBlockObjectWriter(  
2.     blockId: BlockId,  
3.     file: File,  
4.     serializer: Serializer,  
5.     bufferSize: Int,  
6.     compressStream: OutputStream => OutputStream,  
7.     syncWrites: Boolean)  
8.   extends BlockObjectWriter(blockId)  

ShuffleFileGroup：私有内部类，对应了一组shuffle files，每个file对应一个reducer。一个Mapper会分到一个ShuffleFileGroup，把mapper的结果写到这组File里去。

MapStatus

注意到ShuffleMapTask的类型是MapStatus类。MapStatus类是ShuffleMapTask要返回给scheduler的执行结果，包括两个东西：

[java] view
plaincopy

1. class MapStatus(var location: BlockManagerId, var compressedSizes: Array[Byte])  

前者是run这次task的block manager地址（BlockManagerId是一个类，保存了executorId,host, port, nettyPort），后者是output大小，该值会传给接下来的reduce任务。该size是被MapOutputTracker压缩过的。

MapStatus类提供了两个方法如下，ShuffleMapTask进行了复写。

[java] view
plaincopy

1. def writeExternal(out: ObjectOutput) {  
2.   location.writeExternal(out)  
3.   out.writeInt(compressedSizes.length)  
4.   out.write(compressedSizes)  
5. }  
6.   
7. def readExternal(in: ObjectInput) {  
8.   location = BlockManagerId(in)  
9.   compressedSizes = new Array[Byte](in.readInt())  
10.   in.readFully(compressedSizes)  
11. }  

BlockManagerId

BlockManagerId类构造依赖executorId, host, port, nettyPort这些信息。伴生对象维护了一个blockManagerIdCache ，实现为ConcurrentHashMap[BlockManagerId,BlockManagerId]() 。

比如MapStatus的readExternal方法把ObjectInput传入BlockManagerId构造函数的时候，BlockManagerId的apply()方法就会根据ObjectInput取出executorId, host, port,nettyPort信息，把这个BlockManagerIdobj维护到blockManagerIdCache内

ResultTask

构造参数

[java] view
plaincopy

1. private[spark] class ResultTask[T, U](  
2.     stageId: Int,  
3.     var rdd: RDD[T],  
4.     var func: (TaskContext, Iterator[T]) => U,  
5.     _partitionId: Int,  
6.     @transient locs: Seq[TaskLocation],  
7.     var outputId: Int)  
8.   extends Task[U](stageId, _partitionId) with Externalizable {  

ResultTask比较简单，runTask方法调用的是rdd的迭代器：

[java] view
plaincopy

1. override def runTask(context: TaskContext): U = {  
2.   metrics = Some(context.taskMetrics)  
3.   try {  
4.     func(context, rdd.iterator(split, context))  
5.   } finally {  
6.     context.executeOnCompleteCallbacks()  
7.   }  
8. }  

进程模型 vs. 线程模型