1.Shuffle簡介
Shuffle描述着數據從map task輸出到reduce task輸入的這段過程。shuffle是連接Map和Reduce之間的橋梁，Map的輸出要用到Reduce中必須經過shuffle這個環節，shuffle的性能高低直接影響了整個程序的性能和吞吐量。
因為在分布式情況下，reduce task需要跨節點去拉取其它節點上的map task結果。這一過程將會產生網絡資源消耗和內存，磁盤IO的消耗。
通常shuffle分為兩部分：Map階段的數據准備和Reduce階段的數據拷貝處理。一般將在map端的Shuffle稱之為Shuffle Write，在Reduce端的Shuffle稱之為Shuffle Read.
Apache Spark 的 Shuffle 過程與 Apache Hadoop 的 Shuffle 過程有着諸多類似，一些概念可直接套用；
例如，Shuffle 過程中，提供數據的一端，被稱作 Map 端，Map 端每個生成數據的任務稱為 Mapper，對應的，接收數據的一端，被稱作 Reduce 端，Reduce 端每個拉取數據的任務稱為 Reducer；
Shuffle 過程本質上都是將 Map 端獲得的數據使用分區器進行划分，並將數據發送給對應的 Reducer 的過程。
1.1 map端的Shuffle簡述
1)input, 根據split輸入數據，運行map任務;
2)patition, 每個map task都有一個內存緩沖區，存儲着map的輸出結果;
3)spill, 當緩沖區快滿的時候需要將緩沖區的數據以臨時文件的方式存放到磁盤;
4)merge, 當整個map task結束后再對磁盤中這個map task產生的所有臨時文件做合並，生成最終的正式輸出文件，然后等待reduce task來拉數據。
1.2 reduce 端的Shuffle簡述
reduce task在執行之前的工作就是不斷地拉取當前job里每個map task的最終結果，然后對從不同地方拉取過來的數據不斷地做merge，也最終形成一個文件作為reduce task的輸入文件。
1) Copy過程，拉取數據。
2)Merge階段，合並拉取來的小文件
3)Reducer計算
4)Output輸出計算結果
圖形象的描述了MR數據流動的整個過程：
圖解釋：
map端，有4個map;Reduce端，有3個reduce。
4個map 也就是4個JVM，每個JVM處理一個數據分片(split1~split4)，每個map產生一個map輸出文件，但是每個map都為后面的reduce產生了3部分數據(分別用紅1、綠2、藍3標識)，也就是說每個輸出的map文件都包含了3部分數據。
mapper運行后，通過Partitioner接口，根據key或value及reduce的數量來決定當前map的輸出數據最終應該交由哪個reduce task處理.Reduce端一共有3個reduce，去前面的4個map的輸出結果中抓取屬於自己的數據。
2.Spark Shuffle
在Spark的中，負責shuffle過程的執行、計算和處理的組件主要就是ShuffleManager，也即shuffle管理器。ShuffleManager隨着Spark的發展有兩種實現的方式，分別為HashShuffleManager和SortShuffleManager，因此spark的Shuffle有Hash Shuffle和Sort Shuffle兩種
在Spark 1.2以前，默認的shuffle計算引擎是HashShuffleManager。HashShuffleManager有着一個非常嚴重的弊端，就是會產生大量的中間磁盤文件，進而由大量的磁盤IO操作影響了性能。
在Spark 1.2以后，默認的ShuffleManager改成了SortShuffleManager。SortShuffleManager相較於HashShuffleManager來說，有了一定的改進。主要就在於，每個Task在進行shuffle操作時，雖然也會產生較多的臨時磁盤文件，但是最后會將所有的臨時文件合並(merge)成一個磁盤文件，因此每個Task就只有一個磁盤文件。
2.1 Hash shuffle
HashShuffleManager的運行機制主要分成兩種，一種是普通運行機制，另一種是合並的運行機制。
合並機制主要是通過復用buffer來優化Shuffle過程中產生的小文件的數量。Hash shuffle是不具有排序的Shuffle。
2.1.1 普通機制
圖解：
這里我們先明確一個假設前提：每個Executor只有1個CPU core，也就是說，無論這個Executor上分配多少個task線程，同一時間都只能執行一個task線程。
從Map Task 開始，按照 Hash 計算(分區器：hash/numreduce取模)，分類出3個不同的類別，每個 Task 都分成3種類別的數據
每個Reduce Task會在每個Map Task 中把屬於自己類別的數據收集過來，匯聚成一個同類別的大集合
每1個 Task 輸出3份本地文件，這里有4個 Mapper Tasks，所以總共輸出了4個 Tasks x 3個分類文件 = 12個本地小文件。
（1）shuffle write階段
每個 Map task處理的數據按key進行“分區”。所謂“分區”，就是對相同的key執行hash算法，從而將相同key都寫入同一個磁盤文件中，而每一個磁盤文件都只屬於reduce端的stage的一個task。
在將數據寫入磁盤之前，會先將數據寫入內存緩沖中，當內存緩沖填滿之后，才會溢寫到磁盤文件中去。
那么map task，要為下一個stage創建多少個磁盤文件呢?下一個stage的task有多少個，當前stage的每個task就要創建多少份磁盤文件。
（2）shuffle read階段
shuffle read的過程中，每個task只要從上游stage的所有task所在節點上，拉取屬於自己的那一個磁盤文件即可。
shuffle read的拉取過程是一邊拉取一邊進行聚合的。每個shuffle read task都會有一個自己的buffer緩沖，每次都只能拉取與buffer緩沖相同大小的數據，然后通過內存中的一個Map進行聚合等操作。
聚合完一批數據后，再拉取下一批數據，並放到buffer緩沖中進行聚合操作。以此類推，直到最后將所有數據到拉取完，並得到最終的結果。
Hash shuffle普通機制的問題
1).Shuffle前在磁盤上會產生海量的小文件，建立通信和拉取數據的次數變多,此時會產生大量耗時低效的 IO 操作 (因為產生過多的小文件)
2).可能導致OOM，大量耗時低效的 IO 操作 ，導致寫磁盤時的對象過多，讀磁盤時候的對象也過多，這些對象存儲在堆內存中，會導致堆內存不足，相應會導致頻繁的GC，GC會導致OOM。
2.1.1 合並機制
合並機制就是復用buffer，開啟合並機制的配置是spark.shuffle.consolidateFiles。該參數默認值為false，將其設置為true即可開啟優化機制。通常來說，如果我們使用HashShuffleManager，那么都建議開啟這個選項。
這里還是有4個Tasks，數據類別還是分成3種類型，因為Hash算法會根據你的 Key 進行分類，在同一個進程中，無論是有多少過Task，都會把同樣的Key放在同一個Buffer里；
然后把Buffer中的數據寫入以Core數量為單位的本地文件中，(一個Core只有一種類型的Key的數據)，每1個Task所在的進程中，分別寫入共同進程中的3份本地文件；
這里有4個Mapper Tasks，所以總共輸出是 2個Cores x 3個分類文件 = 6個本地小文件。
Hash shuffle合並機制的問題
如果 Reducer 端的並行任務或者是數據分片過多的話則 Core * Reducer Task 依舊過大，也會產生很多小文件。
2.2 Sort shuffle
SortShuffleManager的運行機制主要分成兩種，一種是普通運行機制，另一種是bypass運行機制。
當shuffle read task的數量小於等於spark.shuffle.sort.bypassMergeThreshold參數的值時(默認為200)，就會啟用bypass機制。
2.2.1 普通機制
在該模式下，數據會先寫入一個內存數據結構中(默認5M)，此時根據不同的shuffle算子，可能選用不同的數據結構。
如果是reduceByKey這種聚合類的shuffle算子，那么會選用Map數據結構，一邊通過Map進行聚合，一邊寫入內存;
如果是join這種普通的shuffle算子，那么會選用Array數據結構，直接寫入內存。
接着，每寫一條數據進入內存數據結構之后，就會判斷一下，是否達到了某個臨界閾值。
如果達到臨界閾值的話，那么就會嘗試將內存數據結構中的數據溢寫到磁盤，然后清空內存數據結構。
（1）內存數據
shuffle中的定時器會檢查內存數據結構的大小，如果內存數據結構空間不夠，那么會申請額外的內存，如果申請不到，則發生溢寫。
（2）排序
在溢寫到磁盤文件之前，會先根據key對內存數據結構中已有的數據進行排序。
（3）溢寫
排序過后，會分批將數據寫入磁盤文件。默認的batch數量是10000條，也就是說，排序好的數據，會以每批1萬條數據的形式分批寫入磁盤文件。
寫入磁盤文件是通過Java的BufferedOutputStream實現的。BufferedOutputStream是Java的緩沖輸出流，首先會將數據緩沖在內存中，當內存緩沖滿溢之后再一次寫入磁盤文件中，這樣可以減少磁盤IO次數，提升性能。
（4）merge
一個task將所有數據寫入內存數據結構的過程中，會發生多次磁盤溢寫操作，也就會產生多個臨時文件。最后會將之前所有的臨時磁盤文件都進行合並，這就是merge過程，此時會將之前所有臨時磁盤文件中的數據讀取出來，然后依次寫入最終的磁盤文件之中。
此外，由於一個task就只對應一個磁盤文件，也就意味着該task為Reduce端的stage的task准備的數據都在這一個文件中，因此還會單獨寫一份索引文件，其中標識了下游各個task的數據在文件中的start offset與end offset。
2.2.1 bypass
bypass運行機制的觸發條件如下：
1)shuffle map task數量小於spark.shuffle.sort.bypassMergeThreshold參數的值。
2)不是聚合類的shuffle算子
每個 map task會為每個reduce端的task都創建一個臨時磁盤文件，並將數據按key進行hash然后根據key的hash值，將key寫入對應的磁盤文件之中。
當然，寫入磁盤文件時也是先寫入內存緩沖，緩沖寫滿之后再溢寫到磁盤文件的。
最后，同樣會將所有臨時磁盤文件都合並成一個磁盤文件，並創建一個單獨的索引文件。
而該機制與普通SortShuffleManager運行機制的不同在於：
第一，磁盤寫機制不同;
第二，不會進行排序。也就是說，啟用該機制的最大好處在於，shuffle write過程中，不需要進行數據的排序操作，也就節省掉了這部分的性能開銷。
3. 總結
Shuffle 過程本質上都是將 Map 端獲得的數據使用分區器進行划分，並將數據發送給對應的 Reducer 的過程。
shuffle作為處理連接map端和reduce端的樞紐，其shuffle的性能高低直接影響了整個程序的性能和吞吐量。map端的shuffle一般為shuffle的Write階段，reduce端的shuffle一般為shuffle的read階段。Hadoop和spark的shuffle在實現上面存在很大的不同，spark的shuffle分為兩種實現，分別為HashShuffle和SortShuffle，
HashShuffle又分為普通機制和合並機制，普通機制因為其會產生M*R個數的巨量磁盤小文件而產生大量性能低下的Io操作，從而性能較低，因為其巨量的磁盤小文件還可能導致OOM，HashShuffle的合並機制通過重復利用buffer從而將磁盤小文件的數量降低到Core*R個，但是當Reducer 端的並行任務或者是數據分片過多的時候，依然會產生大量的磁盤小文件。
SortShuffle也分為普通機制和bypass機制，普通機制在內存數據結構(默認為5M)完成排序，會產生2M個磁盤小文件。而當shuffle map task數量小於spark.shuffle.sort.bypassMergeThreshold參數的值。或者算子不是聚合類的shuffle算子(比如reduceByKey)的時候會觸發SortShuffle的bypass機制，SortShuffle的bypass機制不會進行排序，極大的提高了其性能
在Spark 1.2以前，默認的shuffle計算引擎是HashShuffleManager，因為HashShuffleManager會產生大量的磁盤小文件而性能低下，在Spark 1.2以后的版本中，默認的ShuffleManager改成了SortShuffleManager。SortShuffleManager相較於HashShuffleManager來說，有了一定的改進。主要就在於，每個Task在進行shuffle操作時，雖然也會產生較多的臨時磁盤文件，但是最后會將所有的臨時文件合並(merge)成一個磁盤文件，因此每個Task就只有一個磁盤文件。在下一個stage的shuffle read task拉取自己的數據時，只要根據索引讀取每個磁盤文件中的部分數據即可。