SparkShuffle

• SparkShuffle概念

reduceByKey會(huì)將上一個(gè)RDD中的每一個(gè)key對(duì)應(yīng)的所有value聚合成一個(gè)value，然后生成一個(gè)新的RDD，元素類型是<key,value>對(duì)的形式，這樣每一個(gè)key對(duì)應(yīng)一個(gè)聚合起來的value。

問題：聚合之前，每一個(gè)key對(duì)應(yīng)的value不一定都是在一個(gè)partition中，也不太可能在同一個(gè)節(jié)點(diǎn)上，因?yàn)镽DD是分布式的彈性的數(shù)據(jù)集，RDD的partition極有可能分布在各個(gè)節(jié)點(diǎn)上。

如何聚合？

– Shuffle Write：上一個(gè)stage的每個(gè)map task就必須保證將自己處理的當(dāng)前分區(qū)的數(shù)據(jù)相同的key寫入一個(gè)分區(qū)文件中，可能會(huì)寫入多個(gè)不同的分區(qū)文件中。

?– Shuffle Read：reduce task就會(huì)從上一個(gè)stage的所有task所在的機(jī)器上尋找屬于己的那些分區(qū)文件，這樣就可以保證每一個(gè)key所對(duì)應(yīng)的value都會(huì)匯聚到同一個(gè)節(jié)點(diǎn)上去處理和聚合。

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

---

?HashShuffleManager

• 普通機(jī)制
• 普通機(jī)制示意圖

?

• 執(zhí)行流程

每一個(gè)map task將不同結(jié)果寫到不同的buffer中，每個(gè)buffer的大小為32K。buffer起到數(shù)據(jù)緩存的作用。

每個(gè)buffer文件最后對(duì)應(yīng)一個(gè)磁盤小文件。

reduce task來拉取對(duì)應(yīng)的磁盤小文件。

• 總結(jié)

.map task的計(jì)算結(jié)果會(huì)根據(jù)分區(qū)器（默認(rèn)是hashPartitioner）來決定寫入到哪一個(gè)磁盤小文件中去。ReduceTask會(huì)去Map端拉取相應(yīng)的磁盤小文件。

.產(chǎn)生的磁盤小文件的個(gè)數(shù)：

M（map task的個(gè)數(shù)）*R（reduce task的個(gè)數(shù)）

• 存在的問題

? ? ?產(chǎn)生的磁盤小文件過多，會(huì)導(dǎo)致以下問題：

在Shuffle Write過程中會(huì)產(chǎn)生很多寫磁盤小文件的對(duì)象。

在Shuffle Read過程中會(huì)產(chǎn)生很多讀取磁盤小文件的對(duì)象。

在JVM堆內(nèi)存中對(duì)象過多會(huì)造成頻繁的gc,gc還無法解決運(yùn)行所需要的內(nèi)存 的話，就會(huì)OOM。

在數(shù)據(jù)傳輸過程中會(huì)有頻繁的網(wǎng)絡(luò)通信，頻繁的網(wǎng)絡(luò)通信出現(xiàn)通信故障的可能性大大增加，一旦網(wǎng)絡(luò)通信出現(xiàn)了故障會(huì)導(dǎo)致shuffle file cannot find 由于這個(gè)錯(cuò)誤導(dǎo)致的task失敗，TaskScheduler不負(fù)責(zé)重試，由DAGScheduler負(fù)責(zé)重試Stage。

---

• ??合并機(jī)制
• 合并機(jī)制示意圖

• 總結(jié)

產(chǎn)生磁盤小文件的個(gè)數(shù)：C(core的個(gè)數(shù))*R（reduce的個(gè)數(shù)）

---

---

SortShuffleManager

• 普通機(jī)制
• 普通機(jī)制示意圖

?

• 執(zhí)行流程

map task 的計(jì)算結(jié)果會(huì)寫入到一個(gè)內(nèi)存數(shù)據(jù)結(jié)構(gòu)里面，內(nèi)存數(shù)據(jù)結(jié)構(gòu)默認(rèn)是5M

在shuffle的時(shí)候會(huì)有一個(gè)定時(shí)器，不定期的去估算這個(gè)內(nèi)存結(jié)構(gòu)的大小，當(dāng)內(nèi)存結(jié)構(gòu)中的數(shù)據(jù)超過5M時(shí)，比如現(xiàn)在內(nèi)存結(jié)構(gòu)中的數(shù)據(jù)為5.01M，那么他會(huì)申請(qǐng)5.01*2-5=5.02M內(nèi)存給內(nèi)存數(shù)據(jù)結(jié)構(gòu)。

如果申請(qǐng)成功不會(huì)進(jìn)行溢寫，如果申請(qǐng)不成功，這時(shí)候會(huì)發(fā)生溢寫磁盤。

在溢寫之前內(nèi)存結(jié)構(gòu)中的數(shù)據(jù)會(huì)進(jìn)行排序分區(qū)

然后開始溢寫磁盤，寫磁盤是以batch的形式去寫，一個(gè)batch是1萬條數(shù)據(jù)，

map task執(zhí)行完成后，會(huì)將這些磁盤小文件合并成一個(gè)大的磁盤文件，同時(shí)生成一個(gè)索引文件。

reduce task去map端拉取數(shù)據(jù)的時(shí)候，首先解析索引文件，根據(jù)索引文件再去拉取對(duì)應(yīng)的數(shù)據(jù)。

• 總結(jié)

產(chǎn)生磁盤小文件的個(gè)數(shù)： 2*M（map task的個(gè)數(shù)）

---

• ?bypass機(jī)制
• bypass機(jī)制示意圖

?

• 總結(jié)
• .bypass運(yùn)行機(jī)制的觸發(fā)條件如下：

? ? ? ? ? ? shuffle reduce task的數(shù)量小于spark.shuffle.sort.bypassMergeThreshold的參數(shù)值。這個(gè)值默認(rèn)是200。

• .產(chǎn)生的磁盤小文件為：2*M（map task的個(gè)數(shù)）

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。