Spark的shuffle剖析!

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark的shuffle剖析!相关的知识,希望对你有一定的参考价值。

Spark的shuffle剖析!


一、什么是shuffle?

shuffle是洗牌的意思,总的来说,就是分散在各个节点的数据,在经过计算之后,需要重新将数据进行分配,以进行下一步的计算。比如wordcount,显示在3台节点上,分别计算了spark的数量、hadoop的数量、scala的数量,结果如下:


节点1:   (spark 1)   (hadoop 1)

节点2:   (hadoop 1)  (scala 1)

节点3:   (hadoop 1)  (spark 3)

在经过计算之后,下一步就需要汇总了,那么汇总就涉及到shuffle,把数据要分类传输,比如spark的都到节点1,hadoop 的都到节点2,scala的都到节点3。


二、为什么shuffle重要?

Spark大会上,所有的演讲嘉宾都认为shuffle是最影响性能的地方,在MapReduce框架中,shuffle是连接Map和Reduce之间的桥梁,Map的输出要用到Reduce中必须经过shuffle这个环节,shuffle的性能高低直接影响了整个程序的性能和吞吐量。


三、从技术上结识shuffle?


Shuffle是MapReduce框架中的一个特定的phase,介于Map phase和Reduce phase之间,当Map的输出结果要被Reduce使用时,输出结果需要按key哈希,并且分发到每一个Reducer上去,这个过程就是shuffle。由于shuffle涉及到了磁盘的读写和网络的传输,因此shuffle性能的高低直接影响到了整个程序的运行效率。

下面这幅图清晰地描述了MapReduce算法的整个流程,其中shuffle phase是介于Map phase和Reduce phase之间。

技术分享


四、shuffle阶段的划分

Spark的操作模型是基于RDD的,当调用RDD的reduceByKey、groupByKey等类似的操作的时候,就需要有shuffle了。再拿出reduceByKey这个来讲。

def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)] = {
    reduceByKey(new HashPartitioner(numPartitions), func)
  }

reduceByKey的时候,我们可以手动设定reduce的个数,如果不指定的话,就可能不受控制了。

技术分享

def defaultPartitioner(rdd: RDD[_], others: RDD[_]*): Partitioner = {
    val bySize = (Seq(rdd) ++ others).sortBy(_.partitions.size).reverse    for (r <- bySize if r.partitioner.isDefined) {      return r.partitioner.get
    }    if (rdd.context.conf.contains("spark.default.parallelism")) {      new HashPartitioner(rdd.context.defaultParallelism)
    } else {      new HashPartitioner(bySize.head.partitions.size)
    }
  }

View Code

如果不指定reduce个数的话,就按默认的走:

1、如果自定义了分区函数partitioner的话,就按你的分区函数来走。

2、如果没有定义,那么如果设置了 spark.default.parallelism ,就使用哈希的分区方式,reduce个数就是设置的这个值。

3、如果这个也没设置,那就按照输入数据的分片的数量来设定。如果是hadoop的输入数据的话,这个就多了。。。大家可要小心啊。

设定完之后,它会做三件事情,也就是之前讲的3次RDD转换。

//map端先按照key合并一次val combined = self.mapPartitionsWithContext((context, iter) => {  aggregator.combineValuesByKey(iter, context)
 }, preservesPartitioning = true)//reduce抓取数据val partitioned = new ShuffledRDD[K, C, (K, C)](combined, partitioner).setSerializer(serializer)//合并数据,执行reduce计算partitioned.mapPartitionsWithContext((context, iter) => {  new InterruptibleIterator(context, aggregator.combineCombinersByKey(iter, context))
 }, preservesPartitioning = true)

View Code

技术分享

1、在第一个MapPartitionsRDD这里先做一次map端的聚合操作。

2、SHuffledRDD主要是做从这个抓取数据的工作。

3、第二个MapPartitionsRDD把抓取过来的数据再次进行聚合操作。

4、步骤1和步骤3都会涉及到spill的过程。

怎么做的聚合操作,回去看RDD那章。

五、Shuffle的中间结果如何存储

作业提交的时候,DAGScheduler会把Shuffle的过程切分成map和reduce两个Stage(之前一直被我叫做shuffle前和shuffle后),具体的切分的位置在上图的虚线处。

map端的任务会作为一个ShuffleMapTask提交,最后在TaskRunner里面调用了它的runTask方法。

override def runTask(context: TaskContext): MapStatus = {
    val numOutputSplits = dep.partitioner.numPartitions
    metrics = Some(context.taskMetrics)

    val blockManager = SparkEnv.get.blockManager
    val shuffleBlockManager = blockManager.shuffleBlockManager    var shuffle: ShuffleWriterGroup = null
    var success = false

    try {      // serializer为空的情况调用默认的JavaSerializer,也可以通过spark.serializer来设置成别的
      val ser = Serializer.getSerializer(dep.serializer)      // 实例化Writer,Writer的数量=numOutputSplits=前面我们说的那个reduce的数量
      shuffle = shuffleBlockManager.forMapTask(dep.shuffleId, partitionId, numOutputSplits, ser)      // 遍历rdd的元素,按照key计算出来它所在的bucketId,然后通过bucketId找到相应的Writer写入
      for (elem <- rdd.iterator(split, context)) {
        val pair = elem.asInstanceOf[Product2[Any, Any]]
        val bucketId = dep.partitioner.getPartition(pair._1)
        shuffle.writers(bucketId).write(pair)
      }      // 提交写入操作. 计算每个bucket block的大小
      var totalBytes = 0L
      var totalTime = 0L
      val compressedSizes: Array[Byte] = shuffle.writers.map { writer: BlockObjectWriter =>
        writer.commit()
        writer.close()
        val size = writer.fileSegment().length
        totalBytes += size
        totalTime += writer.timeWriting()
        MapOutputTracker.compressSize(size)
      }      // 更新 shuffle 监控参数.
      val shuffleMetrics = new ShuffleWriteMetrics
      shuffleMetrics.shuffleBytesWritten = totalBytes
      shuffleMetrics.shuffleWriteTime = totalTime
      metrics.get.shuffleWriteMetrics = Some(shuffleMetrics)

      success = true
      new MapStatus(blockManager.blockManagerId, compressedSizes)
    } catch { case e: Exception =>      // 出错了,取消之前的操作,关闭writer
      if (shuffle != null && shuffle.writers != null) {        for (writer <- shuffle.writers) {
          writer.revertPartialWrites()
          writer.close()
        }
      }      throw e
    } finally {      // 关闭writer
      if (shuffle != null && shuffle.writers != null) {        try {
          shuffle.releaseWriters(success)
        } catch {          case e: Exception => logError("Failed to release shuffle writers", e)
        }
      }      // 执行注册的回调函数,一般是做清理工作      context.executeOnCompleteCallbacks()
    }
  }

View Code

遍历每一个记录,通过它的key来确定它的bucketId,再通过这个bucket的writer写入数据。

下面我们看看ShuffleBlockManager的forMapTask方法吧。

技术分享

def forMapTask(shuffleId: Int, mapId: Int, numBuckets: Int, serializer: Serializer) = {    new ShuffleWriterGroup {
      shuffleStates.putIfAbsent(shuffleId, new ShuffleState(numBuckets))      private val shuffleState = shuffleStates(shuffleId)      private var fileGroup: ShuffleFileGroup = null

      val writers: Array[BlockObjectWriter] = if (consolidateShuffleFiles) {
        fileGroup = getUnusedFileGroup()        Array.tabulate[BlockObjectWriter](numBuckets) { bucketId =>
          val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
      // 从已有的文件组里选文件,一个bucket一个文件,即要发送到同一个reduce的数据写入到同一个文件          blockManager.getDiskWriter(blockId, fileGroup(bucketId), serializer, bufferSize)
        }
      } else {        Array.tabulate[BlockObjectWriter](numBuckets) { bucketId =>          // 按照blockId来生成文件,文件数为map数*reduce数
          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)
        }
      }


几个重要的结论:

1、map的中间结果是写入到本地硬盘的,而不是内存

2、默认是一个map的中间结果文件是M*R(M=map数量,R=reduce的数量),设置了spark.shuffle.consolidateFiles为true之后是R个文件,根据bucketId把要分到同一个reduce的结果写入到一个文件中。

3、consolidateFiles采用的是一个reduce一个文件,它还记录了每个map的写入起始位置,所以查找的时候,先通过reduceId查找到哪个文件,再同坐mapId查找索引当中的起始位置offset,长度length=(mapId + 1).offset -(mapId).offset,这样就可以确定一个FileSegment(file, offset, length)。

4、Finally,存储结束之后, 返回了一个new MapStatus(blockManager.blockManagerId, compressedSizes),把blockManagerId和block的大小都一起返回。

六、Shuffle的数据如何拉取过来

ShuffleMapTask结束之后,最后在DAGScheduler的handleTaskCompletion方法当中。


1、把结果添加到Stage的outputLocs数组里,它是按照数据的分区Id来存储映射关系的partitionId->MapStaus。

2、stage结束之后,通过mapOutputTracker的registerMapOutputs方法,把此次shuffle的结果outputLocs记录到mapOutputTracker里面。

这个stage结束之后,就到ShuffleRDD运行了,我们看一下它的compute函数。

SparkEnv.get.shuffleFetcher.fetch[P](shuffledId, split.index, context, ser)

它是通过ShuffleFetch的fetch方法来抓取的,具体实现在BlockStoreShuffleFetcher里面。


1、MapOutputTrackerWorker向MapOutputTrackerMaster获取shuffle相关的map结果信息。

2、把map结果信息构造成BlockManagerId --> Array(BlockId, size)的映射关系。

3、通过BlockManager的getMultiple批量拉取block。

4、返回一个可遍历的Iterator接口,并更新相关的监控参数。

我们继续看getMultiple方法。分两种情况处理,分别是netty的和Basic的,Basic的就不讲了,就是通过ConnectionManager去指定的BlockManager那里获取数据,上一章刚好说了。

我们讲一下Netty的吧,这个是需要设置的才能启用的,不知道性能会不会好一些呢?

看NettyBlockFetcherIterator的initialize方法,再看BasicBlockFetcherIterator的initialize方法,发现Basic的不能同时抓取超过48Mb的数据。

技术分享

在NettyBlockFetcherIterator的sendRequest方法里面,发现它是通过ShuffleCopier来试下的。


这块接下来就是netty的客户端调用的方法了,我对这个不了解。在服务端的处理是在DiskBlockManager内部启动了一个ShuffleSender的服务,最终的业务处理逻辑是在FileServerHandler。

它是通过getBlockLocation返回一个FileSegment,下面这段代码是ShuffleBlockManager的getBlockLocation方法。


获取的方法前面说了,通过reduceId找到文件,再通过mapId找到它的起始位置。但是这里有个疑问了,如果启用了consolidateFiles,一个reduce的所需数据都在一个文件里,是不是就可以把整个文件一起返回呢,而不是通过N个map过不需要来分多次读取?还是害怕一次发送一个大文件容易失败?这就不得而知了。

到这里整个过程就讲完了。可以看得出来Shuffle这块还是做了一些优化的,但是这些参数并没有启用,有需要的朋友可以自己启用一下试试效果。




参考资料:

http://blog.csdn.net/hao707822882/article/details/40581515

http://blog.csdn.net/johnny_lee/article/details/22619585


以上是关于Spark的shuffle剖析!的主要内容,如果未能解决你的问题,请参考以下文章

[Spark性能调优] Spark Shuffle 中 JVM 内存使用及配置详情

[Spark性能调优] 第四章 : Spark Shuffle 中 JVM 内存使用及配置内幕详情

spark shuffle写操作三部曲之UnsafeShuffleWriter

spark性能优化:shuffle调优

大数据-spark理论算子,shuffle优化

Spark学习之路 SparkCore的调优之Shuffle调优