SparkShuffle

• SparkShuffle概念

reduceByKey會將上一個RDD中的每一個key對應的所有value聚合成一個value，然后生成一個新的RDD，元素類型是<key,value>對的形式，這樣每一個key對應一個聚合起來的value。

問題：聚合之前，每一個key對應的value不一定都是在一個partition中，也不太可能在同一個節點上，因為RDD是分布式的彈性的數據集，RDD的partition極有可能分布在各個節點上。

如何聚合？

– Shuffle Write：上一個stage的每個map task就必須保證將自己處理的當前分區的數據相同的key寫入一個分區文件中，可能會寫入多個不同的分區文件中。

?– Shuffle Read：reduce task就會從上一個stage的所有task所在的機器上尋找屬于己的那些分區文件，這樣就可以保證每一個key所對應的value都會匯聚到同一個節點上去處理和聚合。

Spark中有兩種Shuffle管理類型，HashShufflManager和SortShuffleManager，Spark1.2之前是HashShuffleManager， Spark1.2引入SortShuffleManager,在Spark 2.0+版本中已經將HashShuffleManager丟棄。

---

?HashShuffleManager

• 普通機制
• 普通機制示意圖

?

• 執行流程

每一個map task將不同結果寫到不同的buffer中，每個buffer的大小為32K。buffer起到數據緩存的作用。

每個buffer文件最后對應一個磁盤小文件。

reduce task來拉取對應的磁盤小文件。

• 總結

.map task的計算結果會根據分區器（默認是hashPartitioner）來決定寫入到哪一個磁盤小文件中去。ReduceTask會去Map端拉取相應的磁盤小文件。

.產生的磁盤小文件的個數：

M（map task的個數）*R（reduce task的個數）

• 存在的問題

? ? ?產生的磁盤小文件過多，會導致以下問題：

在Shuffle Write過程中會產生很多寫磁盤小文件的對象。

在Shuffle Read過程中會產生很多讀取磁盤小文件的對象。

在JVM堆內存中對象過多會造成頻繁的gc,gc還無法解決運行所需要的內存 的話，就會OOM。

在數據傳輸過程中會有頻繁的網絡通信，頻繁的網絡通信出現通信故障的可能性大大增加，一旦網絡通信出現了故障會導致shuffle file cannot find 由于這個錯誤導致的task失敗，TaskScheduler不負責重試，由DAGScheduler負責重試Stage。

---

• ??合并機制
• 合并機制示意圖

• 總結

產生磁盤小文件的個數：C(core的個數)*R（reduce的個數）

---

---

SortShuffleManager

• 普通機制
• 普通機制示意圖

?

• 執行流程

map task 的計算結果會寫入到一個內存數據結構里面，內存數據結構默認是5M

在shuffle的時候會有一個定時器，不定期的去估算這個內存結構的大小，當內存結構中的數據超過5M時，比如現在內存結構中的數據為5.01M，那么他會申請5.01*2-5=5.02M內存給內存數據結構。

如果申請成功不會進行溢寫，如果申請不成功，這時候會發生溢寫磁盤。

在溢寫之前內存結構中的數據會進行排序分區

然后開始溢寫磁盤，寫磁盤是以batch的形式去寫，一個batch是1萬條數據，

map task執行完成后，會將這些磁盤小文件合并成一個大的磁盤文件，同時生成一個索引文件。

reduce task去map端拉取數據的時候，首先解析索引文件，根據索引文件再去拉取對應的數據。

• 總結

產生磁盤小文件的個數： 2*M（map task的個數）

---

• ?bypass機制
• bypass機制示意圖

?

• 總結
• .bypass運行機制的觸發條件如下：

? ? ? ? ? ? shuffle reduce task的數量小于spark.shuffle.sort.bypassMergeThreshold的參數值。這個值默認是200。

• .產生的磁盤小文件為：2*M（map task的個數）

總結

以上是生活随笔為你收集整理的Spark _17 _SparkShuffle、HashShuffleManager、SortShuffleManager的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。