縱觀整個mapreduce過程會發(fā)現(xiàn)存在許多的排序和文件合并操作。
為什么要排序，主要原因有：
1、key的存在combiner操作，排序之后相同的key放到一塊顯然方便做合并操作。
2、reduce task是按key去處理數(shù)據(jù)的。 如果沒有排序那必須從所有數(shù)據(jù)中把當(dāng)前相同key的所有value數(shù)據(jù)拿出來，然后進(jìn)行reduce邏輯處理。顯然每個key到這個邏輯都需要做一次全量數(shù)據(jù)掃描，影響性能，有了排序很方便的得到一個key對于的value集合。
3、reduce task按key去處理數(shù)據(jù)時，如果key按順序排序，那么reduce task就按key順序去讀取，顯然當(dāng)讀到的key是文件末尾的key那么久標(biāo)志數(shù)據(jù)處理完畢。如果沒有排序那還得有其他邏輯來記錄哪些key處理完了，哪些key沒有處理完。
為什么要文件合并，主要原因有：
1、因?yàn)閮?nèi)存放不下就會溢寫文件，就會發(fā)生多次溢寫，形成很多小文件，如果不合并，顯然會小文件泛濫，集群需要資源開銷去管理這些小文件數(shù)據(jù)。
2、任務(wù)去讀取文件的數(shù)增多，打開的文件句柄數(shù)也會增多
3、mapreduce是全局有序。單個文件有序，不代表全局有序，只有把小文件合并一起排序才會全局有序。
雖有千萬種理由需要這么做，但是很耗資源，并且像排序其實(shí)我們有些業(yè)務(wù)并不需要排序。在hadoop 2.x 排序就變?yōu)榭蛇x了。
spark的shuffle是在mapreduce shuffle基礎(chǔ)上進(jìn)行的調(diào)優(yōu)。其實(shí)就是對排序、合并邏輯做了一些優(yōu)化。在spark中shuffle write相當(dāng)于mapreduce 的map,shuffle reade相當(dāng)于mapreduce 的reduce.
spark shuffle分4種
在Spark 1.2以前，默認(rèn)的shuffle計(jì)算引擎是HashShuffleManager
1、未經(jīng)優(yōu)化的HashShuffleManager,其原理見下圖
從圖中可以看到，相比mapreduce，排序不見了，文件合并不見了。上游task寫文件的時候只是將數(shù)據(jù)按分區(qū)追加到文件中，并沒有像mapreduce 那樣先內(nèi)存溢寫成文件，然后再文件與文件之間進(jìn)行合并，雖然節(jié)省了排序、合并的開銷。但有一個弊端就是會產(chǎn)生大量的中間磁盤文件，進(jìn)而由大量的磁盤IO操作影響了性能。如上圖 下游有3個shuffle reade task ,那每個上游shuffle write就會形成3個文件。 形成的文件數(shù)是 shuffle reade個數(shù) × shuffle write個數(shù)。
2、優(yōu)化的HashShuffleManager,其原理見下圖
相比第一種機(jī)制。就是在一個excutor中的task是可以共用一個buffer內(nèi)存。在shuffle write過程中，task就不是為下游stage的每個task創(chuàng)建一個磁盤文件了，而是允許不同的task復(fù)用同一批磁盤文件，這樣就可以有效將多個task的磁盤文件進(jìn)行一定程度上的合并，從而大幅度減少磁盤文件的數(shù)量，進(jìn)而提升shuffle write的性能。此時的文件個數(shù)是 CPU core的數(shù)量? × 下一個stage的task數(shù)量。
為了開啟優(yōu)化后的HashShuffleManager，我們可以設(shè)置一個參數(shù)，spark.shuffle.consolidateFiles。該參數(shù)默認(rèn)值為false，將其設(shè)置為true即可開啟優(yōu)化機(jī)制。通常來說，如果我們使用HashShuffleManager，那么都建議開啟這個選項(xiàng)。
在Spark 1.2以后的版本中，默認(rèn)的ShuffleManager改成了SortShuffleManager
3、SortShuffleManager,其原理見下圖
這種機(jī)制和mapreduce差不多，在該模式下，數(shù)據(jù)會先寫入一個內(nèi)存數(shù)據(jù)結(jié)構(gòu)中，此時根據(jù)不同的shuffle算子，可能選用不同的數(shù)據(jù)結(jié)構(gòu)。如果是reduceByKey這種聚合類的shuffle算子，那么會選用Map數(shù)據(jù)結(jié)構(gòu)，一邊通過Map進(jìn)行聚合，一邊寫入內(nèi)存；如果是join這種普通的shuffle算子，那么會選用Array數(shù)據(jù)結(jié)構(gòu)，直接寫入內(nèi)存。接著，每寫一條數(shù)據(jù)進(jìn)入內(nèi)存數(shù)據(jù)結(jié)構(gòu)之后，就會判斷一下，是否達(dá)到了某個臨界閾值。如果達(dá)到臨界閾值的話，那么就會嘗試將內(nèi)存數(shù)據(jù)結(jié)構(gòu)中的數(shù)據(jù)溢寫到磁盤，然后清空內(nèi)存數(shù)據(jù)結(jié)構(gòu)。
在溢寫到磁盤文件之前，會先根據(jù)key對內(nèi)存數(shù)據(jù)結(jié)構(gòu)中已有的數(shù)據(jù)進(jìn)行排序。排序過后，會分批將數(shù)據(jù)寫入磁盤文件。默認(rèn)的batch數(shù)量是10000條，也就是說，排序好的數(shù)據(jù)，會以每批1萬條數(shù)據(jù)的形式分批寫入磁盤文件。寫入磁盤文件是通過Java的BufferedOutputStream實(shí)現(xiàn)的。BufferedOutputStream是Java的緩沖輸出流，首先會將數(shù)據(jù)緩沖在內(nèi)存中，當(dāng)內(nèi)存緩沖滿溢之后再一次寫入磁盤文件中，這樣可以減少磁盤IO次數(shù)，提升性能。
一個task將所有數(shù)據(jù)寫入內(nèi)存數(shù)據(jù)結(jié)構(gòu)的過程中，會發(fā)生多次磁盤溢寫操作，也就會產(chǎn)生多個臨時文件。最后會將之前所有的臨時磁盤文件都進(jìn)行合并，這就是merge過程，此時會將之前所有臨時磁盤文件中的數(shù)據(jù)讀取出來，然后依次寫入最終的磁盤文件之中。此外，由于一個task就只對應(yīng)一個磁盤文件，也就意味著該task為下游stage的task準(zhǔn)備的數(shù)據(jù)都在這一個文件中，因此還會單獨(dú)寫一份索引文件，其中標(biāo)識了下游各個task的數(shù)據(jù)在文件中的start offset與end offset。
SortShuffleManager由于有一個磁盤文件merge的過程，因此大大減少了文件數(shù)量，由于每個task最終只有一個磁盤文件所以文件個數(shù)等于上游shuffle write個數(shù)。
4、bypass運(yùn)行機(jī)制
相比第3中少了排序，task會為每個下游task都創(chuàng)建一個臨時磁盤文件，并將數(shù)據(jù)按key進(jìn)行hash然后根據(jù)key的hash值，將key寫入對應(yīng)的磁盤文件之中。當(dāng)然，寫入磁盤文件時也是先寫入內(nèi)存緩沖，緩沖寫滿之后再溢寫到磁盤文件的。最后，同樣會將所有臨時磁盤文件都合并成一個磁盤文件，并創(chuàng)建一個單獨(dú)的索引文件。
該過程的磁盤寫機(jī)制其實(shí)跟未經(jīng)優(yōu)化的HashShuffleManager是一模一樣的，因?yàn)槎家獎?chuàng)建數(shù)量驚人的磁盤文件，只是在最后會做一個磁盤文件的合并而已。因此少量的最終磁盤文件，也讓該機(jī)制相對未經(jīng)優(yōu)化的HashShuffleManager來說，shuffle read的性能會更好。
該機(jī)制的最大好處在于，shuffle write過程中，不需要進(jìn)行數(shù)據(jù)的排序操作，也就節(jié)省掉了這部分的性能開銷。
bypass運(yùn)行機(jī)制的觸發(fā)條件如下：
1、shuffle map task數(shù)量小于spark.shuffle.sort.bypassMergeThreshold參數(shù)的值。
2、不是聚合類的shuffle算子（比如reduceByKey）。因?yàn)椴幌竦?種機(jī)制那樣會對聚合類算子以map的數(shù)據(jù)結(jié)構(gòu)存儲，在寫的過程中會先進(jìn)行局部聚合。
spark shuffle 優(yōu)于mapreduce shuffle的原因1、減少了磁盤io
2、可選的shuffle和排序
