在Spark中，什么情况下，会发生shuffle？reduceByKey、groupByKey、sortByKey、countByKey、join、cogroup等操作
Spark Shuffle操作的两个特点
第一个特点 在Spark早期版本中，那个bucket缓存是非常非常重要的，因为需要将一个ShuffleMapTask所有的数据都写入内存缓存之后，才会刷新到磁盘。但是这就有一个问题，如果map side数据过多，那么很容易造成内存溢出。所以spark在新版本中，优化了，默认那个内存缓存是100kb，然后呢，写入一点数据达到了刷新到磁盘的阈值之后，就会将数据一点一点地刷新到磁盘。这种操作的优点，是不容易发生内存溢出。缺点在于，如果内存缓存过小的话，那么可能发生过多的磁盘写io操作。所以，这里的内存缓存大小，是可以根据实际的业务情况进行优化的。第二个特点， 与MapReduce完全不一样的是，MapReduce它必须将所有的数据都写入本地磁盘文件以后，才能启动reduce操作，来拉取数据。为什么？因为mapreduce要实现默认的根据key的排序！所以要排序，肯定得写完所有数据，才能排序，然后reduce来拉取。但是Spark不需要，spark默认情况下，是不会对数据进行排序的。因此ShuffleMapTask每写入一点数据，ResultTask就可以拉取一点数据，然后在本地执行我们定义的聚合函数和算子，进行计算。spark这种机制的好处在于，速度比mapreduce快多了。但是也有一个问题，mapreduce提供的reduce，是可以处理每个key对应的value上的，很方便。但是spark中，由于这种实时拉取的机制，因此提供不了，直接处理key对应的values的算子，只能通过groupByKey，先shuffle，有一个MapPartitionsRDD，然后用map算子，来处理每个key对应的values。就没有mapreduce的计算模型那么方便。
普通的Shuffle操作原理剖析
每个ShuffleMapTask，都会为每个ResultTask创建一份bucket缓存，以及对应的ShuffleBlockFile磁盘文件
所以假设，如果有100个ShuffleMapTask，100个ResultTask，本地磁盘要产生10000个文件，磁盘IO过多，影响性能
ShuffleMapTask的输出，会作为MapStatus，发送到DAGScheduler的MapOutputTrackerMaster中，MapStatus包含了每个ResultTask要拉取的数据大大小
每个ResultTask会用BlockStoreShuffleFetcher去MapOutputTrackerMaster获取自己的要拉取的文件的信息，然后底层通过BlockManager将数据拉取过来
Map端的数据，可以理解为Shuffle的第一个RDD，MapPartitionsRDD 每个ResultTask拉取过来的数据，其实会组成一个内部的RDD，叫ShuffledRDD，优先放入内存，如果内存不够，那么写入磁盘
然后每个ResultTask针对数据进行聚合，最后生成MapPartitionsRDD，就是我们执行reduceByKey等操作希望获得的那个RDD
优化后的Shuffle原理剖析
Spark新版本中，引入了consolidation机制，也就说，提出了ShuffleGroup概念
一个ShuffleMapTask将数据写入ResultTask数量的本地文件，这个不会变，但是，当下一个ShuffleMapTask运行的时候，可以直接将数据写入之前的ShuffleMapTask的本地文件，相当于是，对多个ShuffleMapTask的输出进行了合并，从而大大减少了本地磁盘的数量
一组ShuffleGroup，每个文件中，都存储了多个ShuffleMapTask的数据，每个ShuffleMapTask的数据，叫做一个segment。此外，会通过一些索引，来标记每个ShuffleMapTask的输出在ShuffleBlockFile中的索引，以及偏移量等，来进行不同ShuffleMapTask的数据的区分
机器上，有两个cpu，也就说，4个ShuffleMapTask，有2个ShuffleMapTask是可以并行执行的，先执行的是黑色边框的两个ShuffleMapTask，再执行红色边框的两个ShuffleMapTask 并行执行的ShuffleMapTask，写入的文件，一定是不同的，当一批并行执行的ShuffleMapTask运行完之后，那么新的一批ShuffleMapTask启动起来并行执行的时候，优化机制就开始发挥作用了，开启优化Shuffle操作，主要是通过在SparkConf中，设置一个参数就可以 每一个节点上面，比如有2个cpu，100个ShuffleMapTask，那么此时，就会产生100*100个磁盘文件，就是10000个 开启了consolidation机制之后的Shuffle write ,每个节点上的磁盘文件，数量变成了cpu core的数量 * ShuffleMapTask的数量，同样的情况，每个节点只产生200个磁盘文件
源码
Shuffle写
入口
// 这个writer默认是HashShuffleWriter
writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
看HashShuffleWriter的 writer
方法
// 将每个ShuffleMapTask计算出来的新的rdd的partition数据，写入本地磁盘
override def write(records: Iterator[_ <: Product2[K, V]]): Unit = {
// 首先判断，是否需要在map端本地进行聚合，这里的话，如果是reduceByKey这种操作，它的dep.aggregator.isDefined、dep.mapSideCombine都是true
// 那么就会进行map端的本地聚合
val iter = if (dep.aggregator.isDefined) {
if (dep.mapSideCombine) {
// 这里就会执行本地聚合，比如本地有(hello，1)、(hello，1)，那么此时就会聚合成(hello，2)
dep.aggregator.get.combineValuesByKey(records, context)
} else {
records
}
} else {
require(!dep.mapSideCombine, "Map-side combine without Aggregator specified!")
records
}
// 需要本地聚合，那么先本地聚合，然后遍历数据，对每个数据，调用partitioner，默认是HashPartitioner
// 生成bucketId，也就是决定了，每一份数据，要写入哪个bucket中
for (elem <- iter) {
val bucketId = dep.partitioner.getPartition(elem._1)
// 获取到了bucketId之后，会调用shuffleBlockManager.forMapTask()方法，来生成bucketId对应的writer，然后用writer将数据写入bucket
shuffle.writers(bucketId).write(elem)
}
}
上面代码的shuffle其实是
private val shuffle = shuffleBlockManager.forMapTask(dep.shuffleId, mapId, numOutputSplits, ser,
writeMetrics)
看看 forMapTask()
方法
// 给每一个mao task 获取一个ShuffleWriterGroup
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
// 对应之前讲解的，shuffle有两种模式，一种是普通的，一种是优化后的
// 这里会判断，如果开启了consolidation机制，也就是consolidateShuffleFiles为true的话，那么实际上，不会给每个bucket都获取一个独立的文件
// 而是为这个bucket获取一个ShuffleGroup的writer
val writers: Array[BlockObjectWriter] = if (consolidateShuffleFiles) {
fileGroup = getUnusedFileGroup()
Array.tabulate[BlockObjectWriter](numBuckets) { bucketId =>
// 首先，用shuffleId、mapId、bucketId(也就是reduceId，一个bucket对应一个reduce)生成一个唯一的ShuffleBlockId
// 然后用buckId，来调用ShuffleFileGroup的apply()函数，为bucket获取一个ShuffleFileGroup
val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
// 然后调用BlockManager的getDiskWriter()方法，针对ShuffleFileGroup获取一个writer
// 这里，就明白了，如果开启了consolidation机制，实际上，对于每一个bucket，都会获取一个针对ShuffleFileGroup的wtriter，而不是一个独立的ShuffleBlockFile的writer
// 这样就实现了所谓的，多个ShuffleMapTask的输出数据的合并
blockManager.getDiskWriter(blockId, fileGroup(bucketId), serializer, bufferSize,
writeMetrics)
}
} else {
// 如果没有开启consolidation机制，也就是普通的shuffle操作的话
Array.tabulate[BlockObjectWriter](numBuckets) { bucketId =>
// 同样生成一个ShuffleBlockId
val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
// 然后调用BlockManager的diskBlockManager，获取一个代表了要写入本地磁盘文件的blockFile
val blockFile = blockManager.diskBlockManager.getFile(blockId)
// Because of previous failures, the shuffle file may already exist on this machine.
// If so, remove it.
// 而且会判断，如果blockFile要是存在的话，还得删除它
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()方法，针对那个blockFile生成writer
blockManager.getDiskWriter(blockId, blockFile, serializer, bufferSize, writeMetrics)
}
// 所以使用这种普通的shuffle操作的话，对于每一个ShuffleMapTask输出的bucket，都会在本地获取一个单独的ShuffleBlockFile
}
shuffle读
入口
override def compute(split: Partition, context: TaskContext): Iterator[(K, C)] = {
// ResultTask或者ShuffleMapTask在执行到ShuffleRDD时，肯定会调用ShuffleRDD的computer()方法，来计算当前这个RDD的partition的数据
// 在这里，会调用ShuffleManager的getReader()方法，获取一个HashShuffleReader，然后调用它的read()方法，拉取该ResultTask ShuffleMapTask需要聚合的数据
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)]]
}
看看HashShuffleReader的 read()
方法
/** Read the combined key-values for this reduce task */
override def read(): Iterator[Product2[K, C]] = {
val ser = Serializer.getSerializer(dep.serializer)
// reduceTask在拉取数据时，其实会用BlockStoreShuffleFetcher来从DAGDcheduler的MapOutputTrackerMaster中获取自己想要的数据的信息
// 然后底层，再通过blockManager从对应的位置，拉取需要的数据
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 {
require(!dep.mapSideCombine, "Map-side combine without Aggregator specified!")
// Convert the Product2s to pairs since this is what downstream RDDs currently expect
iter.asInstanceOf[Iterator[Product2[K, C]]].map(pair => (pair._1, pair._2))
}
// Sort the output if there is a sort ordering defined.
dep.keyOrdering match {
case Some(keyOrd: Ordering[K]) =>
// Create an ExternalSorter to sort the data. Note that if spark.shuffle.spill is disabled,
// the ExternalSorter won't spill to disk.
val sorter = new ExternalSorter[K, C, C](ordering = Some(keyOrd), serializer = Some(ser))
sorter.insertAll(aggregatedIter)
context.taskMetrics.incMemoryBytesSpilled(sorter.memoryBytesSpilled)
context.taskMetrics.incDiskBytesSpilled(sorter.diskBytesSpilled)
sorter.iterator
case None =>
aggregatedIter
}
}
看看 BlockStoreShuffleFetcher.fetch()
方法
private[hash] object BlockStoreShuffleFetcher extends Logging {
def fetch[T](
shuffleId: Int,
reduceId: Int,
context: TaskContext,
serializer: Serializer)
: Iterator[T] =
{
logDebug("Fetching outputs for shuffle %d, reduce %d".format(shuffleId, reduceId))
// 拿到全局的blockManager
val blockManager = SparkEnv.get.blockManager
val startTime = System.currentTimeMillis
// 拿到一个全局的MapOutputTracker的引用，调用其getServerStatuses()方法，传入了shuffleId和reduceId
// shuffleId可以代表当前这个stage的上一个stage，shuffle是分为两个stage的，shuffle writer发生在上一个stage中，shuffle read 是发生在当前stage中的
// 首先通过shuffleId可以限制到上一个stage的所有ShuffleMapTask的输出的MapStatus，接着，通过reduceId，也就是所谓的bucketId，来限制，从每个mapTask中，获取当前这个resultTask需要获取的每个ShuffleMapTask的输出文件的信息
// getServerStatuses()方法，一定是走远程网络通信的，因为要联系Driver上的DAGScheduler的MapOutputTrackerMaster
val statuses = SparkEnv.get.mapOutputTracker.getServerStatuses(shuffleId, reduceId)
logDebug("Fetching map output location for shuffle %d, reduce %d took %d ms".format(
shuffleId, reduceId, System.currentTimeMillis - startTime))
// 对刚才拉取的数据，进行一些数据结构上的转换操作
val splitsByAddress = new HashMap[BlockManagerId, ArrayBuffer[(Int, Long)]]
for (((address, size), index) <- statuses.zipWithIndex) {
splitsByAddress.getOrElseUpdate(address, ArrayBuffer()) += ((index, size))
}
val blocksByAddress: Seq[(BlockManagerId, Seq[(BlockId, Long)])] = splitsByAddress.toSeq.map {
case (address, splits) =>
(address, splits.map(s => (ShuffleBlockId(shuffleId, s._1, reduceId), s._2)))
}
def unpackBlock(blockPair: (BlockId, Try[Iterator[Any]])) : Iterator[T] = {
val blockId = blockPair._1
val blockOption = blockPair._2
blockOption match {
case Success(block) => {
block.asInstanceOf[Iterator[T]]
}
case Failure(e) => {
blockId match {
case ShuffleBlockId(shufId, mapId, _) =>
val address = statuses(mapId.toInt)._1
throw new FetchFailedException(address, shufId.toInt, mapId.toInt, reduceId, e)
case _ =>
throw new SparkException(
"Failed to get block " + blockId + ", which is not a shuffle block", e)
}
}
}
}
// ShuffleBlockFetcherIterator构造以后，在其内部，就直接根据拉取到的地理位置信息，通过BlockManager去远程ShuffleMapTask所在的节点的BlockManager去拉取数据
val blockFetcherItr = new ShuffleBlockFetcherIterator(
context,
SparkEnv.get.blockManager.shuffleClient,
blockManager,
blocksByAddress,
serializer,
SparkEnv.get.conf.getLong("spark.reducer.maxMbInFlight", 48) * 1024 * 1024)
val itr = blockFetcherItr.flatMap(unpackBlock)
// 最后，将拉取到的数据，执行一些转换和封装，返回
val completionIter = CompletionIterator[T, Iterator[T]](itr, {
context.taskMetrics.updateShuffleReadMetrics()
})
new InterruptibleIterator[T](context, completionIter) {
val readMetrics = context.taskMetrics.createShuffleReadMetricsForDependency()
override def next(): T = {
readMetrics.incRecordsRead(1)
delegate.next()
}
}
}
}
