spark中shuffle的過程------不看你後悔

阿新 • • 發佈：2019-01-09

Spark大會上，所有的演講嘉賓都認為shuffle是最影響效能的地方，但是又無可奈何。之前去百度面試hadoop的時候，也被問到了這個問題，直接回答了不知道。

這篇文章主要是沿著下面幾個問題來開展：

1、shuffle過程的劃分？

2、shuffle的中間結果如何儲存？

3、shuffle的資料如何拉取過來？

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)
        }
      }

View Code

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這塊和hadoop的機制差別不大，tez這樣的引擎會趕上spark的速度呢？還是讓我們拭目以待吧！

Shuffle的資料如何拉取過來

ShuffleMapTask結束之後，最後在DAGScheduler的handleTaskCompletion方法當中。

case smt: ShuffleMapTask =>
val status = event.result.asInstanceOf[MapStatus]
val execId = status.location.executorId
if (failedEpoch.contains(execId) && smt.epoch <= failedEpoch(execId)) {
logInfo("Ignoring possibly bogus ShuffleMapTask completion from " + execId)
} else {
stage.addOutputLoc(smt.partitionId, status)
}
if (runningStages.contains(stage) && pendingTasks(stage).isEmpty) {
markStageAsFinished(stage)
if (stage.shuffleDep.isDefined) {
 // 真的map過程才會有這個依賴，reduce過程None
 mapOutputTracker.registerMapOutputs(
   　　stage.shuffleDep.get.shuffleId,
 stage.outputLocs.map(list => if (list.isEmpty) null else list.head).toArray,
 changeEpoch = true)
 }
  clearCacheLocs()
  if (stage.outputLocs.exists(_ == Nil)) {
  // 一些任務失敗了，需要重新提交stage
  submitStage(stage)
   } else {
  // 提交下一批任務			  
　　　}
}

View Code

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裡面。

override def fetch[T](
      shuffleId: Int,
      reduceId: Int,
      context: TaskContext,
      serializer: Serializer)
    : Iterator[T] =
{
    val blockManager = SparkEnv.get.blockManager
    val startTime = System.currentTimeMillis
　　 // mapOutputTracker也分Master和Worker，Worker向Master請求獲取reduce相關的MapStatus，主要是（BlockManagerId和size）
    val statuses = SparkEnv.get.mapOutputTracker.getServerStatuses(shuffleId, reduceId)
    // 一個BlockManagerId對應多個檔案的大小
    val splitsByAddress = new HashMap[BlockManagerId, ArrayBuffer[(Int, Long)]]
    for (((address, size), index) <- statuses.zipWithIndex) {
      splitsByAddress.getOrElseUpdate(address, ArrayBuffer()) += ((index, size))
    }
    // 構造BlockManagerId 和 BlockId的對映關係，想不到ShffleBlockId的mapId，居然是1,2,3,4的序列...
    val blocksByAddress: Seq[(BlockManagerId, Seq[(BlockId, Long)])] = splitsByAddress.toSeq.map {
      case (address, splits) =>
        (address, splits.map(s => (ShuffleBlockId(shuffleId, s._1, reduceId), s._2)))
    }
    // 名為updateBlock，實際是檢驗函式，每個Block都對應著一個Iterator介面，如果該介面為空，則應該報錯
    def unpackBlock(blockPair: (BlockId, Option[Iterator[Any]])) : Iterator[T] = {
      val blockId = blockPair._1
      val blockOption = blockPair._2
      blockOption match {
        case Some(block) => {
          block.asInstanceOf[Iterator[T]]
        }
        case None => {
          blockId match {
            case ShuffleBlockId(shufId, mapId, _) =>
              val address = statuses(mapId.toInt)._1
              throw new FetchFailedException(address, shufId.toInt, mapId.toInt, reduceId, null)
            case _ =>
              throw new SparkException("Failed to get block " + blockId + ", which is not a shuffle block")
          }
        }
      }
    }
    // 從blockManager獲取reduce所需要的全部block，並新增校驗函式
    val blockFetcherItr = blockManager.getMultiple(blocksByAddress, serializer)
    val itr = blockFetcherItr.flatMap(unpackBlock)
    
　　val completionIter = CompletionIterator[T, Iterator[T]](itr, {
      // CompelteIterator迭代結束之後，會執行以下這部分程式碼，提交它記錄的各種引數
      val shuffleMetrics = new ShuffleReadMetrics
      shuffleMetrics.shuffleFinishTime = System.currentTimeMillis
      shuffleMetrics.fetchWaitTime = blockFetcherItr.fetchWaitTime
      shuffleMetrics.remoteBytesRead = blockFetcherItr.remoteBytesRead
      shuffleMetrics.totalBlocksFetched = blockFetcherItr.totalBlocks
      shuffleMetrics.localBlocksFetched = blockFetcherItr.numLocalBlocks
      shuffleMetrics.remoteBlocksFetched = blockFetcherItr.numRemoteBlocks
      context.taskMetrics.shuffleReadMetrics = Some(shuffleMetrics)
    })

    new InterruptibleIterator[T](context, completionIter)
  }
}

View Code

1、MapOutputTrackerWorker向MapOutputTrackerMaster獲取shuffle相關的map結果資訊。

2、把map結果資訊構造成BlockManagerId --> Array(BlockId, size)的對映關係。

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

4、返回一個可遍歷的Iterator介面，並更新相關的監控引數。

我們繼續看getMultiple方法。

  def getMultiple(
      blocksByAddress: Seq[(BlockManagerId, Seq[(BlockId, Long)])],
      serializer: Serializer): BlockFetcherIterator = {
    val iter =
      if (conf.getBoolean("spark.shuffle.use.netty", false)) {
        new BlockFetcherIterator.NettyBlockFetcherIterator(this, blocksByAddress, serializer)
      } else {
        new BlockFetcherIterator.BasicBlockFetcherIterator(this, blocksByAddress, serializer)
      }

    iter.initialize()
    iter
  }

View Code

分兩種情況處理，分別是netty的和Basic的，Basic的就不講了，就是通過ConnectionManager去指定的BlockManager那裡獲取資料，上一章剛好說了。

我們講一下Netty的吧，這個是需要設定的才能啟用的，不知道效能會不會好一些呢？

看NettyBlockFetcherIterator的initialize方法，再看BasicBlockFetcherIterator的initialize方法，發現Basic的不能同時抓取超過48Mb的資料。

override def initialize() {
// 分開本地請求和遠端請求，返回遠端的FetchRequest
val remoteRequests = splitLocalRemoteBlocks()
// 抓取順序隨機
for (request <- Utils.randomize(remoteRequests)) {
  fetchRequestsSync.put(request)
}
// 預設是開6個執行緒去進行抓取
copiers = startCopiers(conf.getInt("spark.shuffle.copier.threads", 6))// 讀取本地的block
getLocalBlocks()
   }

View Code

在NettyBlockFetcherIterator的sendRequest方法裡面，發現它是通過ShuffleCopier來試下的。

val cpier = new ShuffleCopier(blockManager.conf)
      cpier.getBlocks(cmId, req.blocks, putResult)

這塊接下來就是netty的客戶端呼叫的方法了，我對這個不瞭解。在服務端的處理是在DiskBlockManager內部啟動了一個ShuffleSender的服務，最終的業務處理邏輯是在FileServerHandler。

它是通過getBlockLocation返回一個FileSegment，下面這段程式碼是ShuffleBlockManager的getBlockLocation方法。

def getBlockLocation(id: ShuffleBlockId): FileSegment = {
// Search all file groups associated with this shuffle.
val shuffleState = shuffleStates(id.shuffleId)
for (fileGroup <- shuffleState.allFileGroups) {
  val segment = fileGroup.getFileSegmentFor(id.mapId, id.reduceId)
  if (segment.isDefined) { return segment.get }
}
throw new IllegalStateException("Failed to find shuffle block: " + id)
  }

獲取的方法前面說了，通過reduceId找到檔案，再通過mapId找到它的起始位置。但是這裡有個疑問了，如果啟用了consolidateFiles，一個reduce的所需資料都在一個檔案裡，是不是就可以把整個檔案一起返回呢，而不是通過N個map過不需要來分多次讀取？還是害怕一次傳送一個大檔案容易失敗？這就不得而知了。

到這裡整個過程就講完了。可以看得出來Shuffle這塊還是做了一些優化的，但是這些引數並沒有啟用，有需要的朋友可以自己啟用一下試試效果。

岑玉海

轉載請註明出處，謝謝！

spark中shuffle的過程------不看你後悔

Shuffle過程的劃分

Shuffle的中間結果如何儲存

Shuffle的資料如何拉取過來

spark中shuffle的過程------不看你後悔

提高程式併發量的幾個建議(不看保證後悔死你)

[乾貨]總結的真好，Android重點知識點總結，不看你會後悔！

【Spark篇】---Spark中Shuffle文件的尋址

MapReduce中shuffle過程

MapReduce和spark的shuffle過程詳解

關於centos完美安裝MsSql資料庫！！！（不看肯定後悔的一定後悔！！！！

KDD 2018 | Airbnb用機器學習實現房屋動態定價（不看你就out啦！）

史上最全的資料庫面試題，不看絕對後悔

面試時95%會被問道的問題，不看會後悔【轉】

標籤用法，不看會後悔！

【轉】網際網路行業各大崗位分析，全是乾貨，不看你就OUT了

Java歷史簡介-不看會後悔

37 個 MySQL 資料庫小技巧，不看別後悔！

這是一篇對Python深入解析的文章，附帶入門技巧，不看你就虧了！

Spark的Shuffle過程

剖析Hadoop和Spark的Shuffle過程差異（二）

《一頭扎進》系列之Python+Selenium框架設計篇1- 價值好幾K的框架，不看別後悔，過時不候

spring bean的單例和多例的使用場景和在單例bean中注入多例（不看後悔，一看必懂）

mysql中char和varchar和text的區別（不看後悔，一看必懂）

spark中shuffle的過程------不看你後悔

Shuffle過程的劃分

Shuffle的中間結果如何儲存

Shuffle的資料如何拉取過來

相關推薦