概述

本文介紹RDD的Shuffle原理，並分析shuffle過程的實現。

RDD Shuffle簡介

spark的某些操作會觸發被稱為shuffle的事件。shuffle是Spark重新分配資料的機制，它可以對資料進行分組，該操作可以跨不同分割槽。該操作通常會在不同的執行器(executor)和主機之間複製資料，這使shuffle成為複雜且非常消耗資源的操作。

Shuffle背景

為了理解shuffle過程，我們可以拿reduceByKey操作進行舉例。reduceByKey操作會產生一個新的RDD，其中單個鍵的所有值都組合成一個元組(tuple) - 對key和與該key關聯的所有值執行reduce函式的結果。挑戰在於，並非單個key的所有值都必須位於同一個分割槽，甚至是同一個機器上，但它們必須能夠被定位到才能計算結果。

在Spark中，資料通常不跨分割槽分佈，以便特定操作能夠訪問到必要位置。

在計算過程中，單個任務將在單個分割槽上執行 - 因此，要組織單個reduceByKey reduce任務執行的所有資料，Spark需要執行all-to-all的操作。Spark必須從所有分割槽讀取以查詢所有key的所有值，然後將各個值組合在一起以計算每個key的最終結果 - 這個過程稱為shuffle。

雖然新shuffle資料的每個分割槽中的元素集是確定的，且分割槽本身的順序也是確定的，但分割槽中的資料的順序是不確定的。如果在shuffle後希望得到特定順序的資料，則可以使用：

• 在mapPartitions使用例如.sorted對每個分割槽進行排序
• repartitionAndSortWithinPartitions在同時重新分割槽的同時有效地對分割槽進行排序
• sortBy來建立一個全域性排序的RDD

可能導致shuffle的操作包括以下幾種：

• 重新分割槽(repartition)操作，例如: repartition和 coalesce。
• ByKey操作（計數除外），例如：如groupByKey和reduceByKey
• 聯合(join)操作，例如：cogroup和join。

shuffle對效能的影響

shuffle操作十分昂貴(消耗效能和資源)，因為它包括：磁碟I/O，資料序列號，網路I/O等操作。為了對數進行shuffle，Spark建立了一個任務集，map任務負責組織資料，reduce任務負責對資料進行聚合。這兩個術語來自MapReduce，但與Spark的map和reduce操作並沒有直接關係。

在內部，各個map任務的結果會保留在記憶體中，直到它們不再合適(fit)。然後，這些結果基於目標分割槽進行排序並寫入單個檔案。reduce任務讀取相關的排好序的資料塊。

某些shuffle操作會消耗大量的堆記憶體，因為它們使用記憶體中的資料結構來在傳輸記錄之前或之後組織記錄。具體來說，reduceByKey和aggregateByKey在map端建立這些結構，並且’ByKey操作在reduce端生成這些結構。當資料不適合記憶體時，Spark會將這些寫入到磁碟，從而導致磁碟I / O的額外開銷和垃圾收集的增加。

Shuffle還會在磁碟上生成大量中間檔案。從Spark 1.3開始，這些檔案將被保留，直到對應的RDD不再被使用，且被垃圾回收。這樣做是為了在重新計算譜系時不需要重新建立shuffle檔案。如果應用程式保留對這些RDD的引用或GC不經常啟動，則垃圾收集可能僅在很長一段時間後才會發生。這意味著長時間執行的Spark作業可能會佔用大量磁碟空間。配置Spark上下文時，spark.local.dir配置引數指定臨時儲存目錄。

可以通過調整各種配置引數來調整shuffle行為。具體的配置將會在實戰部分講到。

注：以上兩端來自官方文件對shuffle過程的說明。

理解shuffle過程

從以上分析我們可以看出，分割槽之間的物理資料的搬遷被稱為shuffling。當需要構建新的RDD時，為了構建新的分割槽需要整合多個分割槽的資料，就會發生shuffling。例如：當通過key對成員進行分組時，Spark可能會掃描所有分割槽，來發現具有相同key的元素，然後在物理上對這些資料進行分組。

從以下圖中可以看出具體shuffling過程：

上圖是《spark in action》中的例子圖。

上圖是一個通過key來聚合資料的例子，開始的時候各個key的資料分佈在不同的Spark節點上。先進行transform過程，轉換過程是在各個資料的分割槽上進行，聚合過程需要從不同節點的分割槽上獲取資料，此時將會發生shuffling。

注意：關於shuffle的詳細程式碼實現，還會有文章專門進行講解。

參考資料

• spark官方文件