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.
Hadoop MapReduce Shuffle
Apache Spark 的 Shuffle 過程與 Apache Hadoop 的 Shuffle 過程有着諸多類似，一些概念可直接套用，例如，Shuffle 過程中，提供數據的一端，被稱作 Map 端，Map 端每個生成數據的任務稱爲 Mapper，對應的，接收數據的一端，被稱作 Reduce 端，Reduce 端每個拉取數據的任務稱爲 Reducer，Shuffle 過程本質上都是將 Map 端獲得的數據使用分區器進行劃分，並將數據發送給對應的 Reducer 的過程。
map端的Shuffle簡述:
1)input, 根據split輸入數據，運行map任務;
2)patition, 每個map task都有一個內存緩衝區，存儲着map的輸出結果;
3)spill, 當緩衝區快滿的時候需要將緩衝區的數據以臨時文件的方式存放到磁盤;
4)merge, 當整個map task結束後再對磁盤中這個map task產生的所有臨時文件做合併，生成最終的正式輸出文件，然後等待reduce task來拉數據。
reduce 端的Shuffle簡述:
reduce task在執行之前的工作就是不斷地拉取當前job裏每個map task的最終結果，然後對從不同地方拉取過來的數據不斷地做merge，也最終形成一個文件作爲reduce task的輸入文件。
1) Copy過程，拉取數據。
2)Merge階段，合併拉取來的小文件
3)Reducer計算
4)Output輸出計算結果
我們可以將Shuffle的過程以數據流的方式呈現：
圖形象的描述了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的輸出結果中抓取屬於自己的數據。
關於Hadoop MR的Shuffle的詳細請查看博客：「戲」說hadoop--hadoop MapReduce Shuffle過程詳解
Spark Shuffle
在「戲」說Spark-Spark核心-Stage劃分及Pipline的計算模式一文中，我們知道stage中是高效快速的pipline的計算模式，寬依賴之間會劃分stage，而Stage之間就是Shuffle，如圖中的stage0，stage1和stage3之間就會產生Shuffle。
在Spark的中，負責shuffle過程的執行、計算和處理的組件主要就是ShuffleManager，也即shuffle管理器。ShuffleManager隨着Spark的發展有兩種實現的方式，分別爲HashShuffleManager和SortShuffleManager，因此spark的Shuffle有Hash Shuffle和Sort Shuffle兩種
Spark Shuffle發展史
在Spark的版本的發展，ShuffleManager在不斷迭代，變得越來越先進。
在Spark 1.2以前，默認的shuffle計算引擎是HashShuffleManager。該ShuffleManager而HashShuffleManager有着一個非常嚴重的弊端，就是會產生大量的中間磁盤文件，進而由大量的磁盤IO操作影響了性能。因此在Spark 1.2以後的版本中，默認的ShuffleManager改成了SortShuffleManager。SortShuffleManager相較於HashShuffleManager來說，有了一定的改進。主要就在於，每個Task在進行shuffle操作時，雖然也會產生較多的臨時磁盤文件，但是最後會將所有的臨時文件合併(merge)成一個磁盤文件，因此每個Task就只有一個磁盤文件。在下一個stage的shuffle read task拉取自己的數據時，只要根據索引讀取每個磁盤文件中的部分數據即可。
Hash shuffle
HashShuffleManager的運行機制主要分成兩種，一種是普通運行機制，另一種是合併的運行機制。
合併機制主要是通過複用buffer來優化Shuffle過程中產生的小文件的數量。Hash shuffle是不具有排序的Shuffle。
普通機制的Hash shuffle
圖解：
這裏我們先明確一個假設前提：每個Executor只有1個CPU core，也就是說，無論這個Executor上分配多少個task線程，同一時間都只能執行一個task線程。
圖中有3個 Reducer，從Task 開始那邊各自把自己進行 Hash 計算(分區器：hash/numreduce取模)，分類出3個不同的類別，每個 Task 都分成3種類別的數據，想把不同的數據匯聚然後計算出最終的結果，所以Reducer 會在每個 Task 中把屬於自己類別的數據收集過來，匯聚成一個同類別的大集合，每1個 Task 輸出3份本地文件，這裏有4個 Mapper Tasks，所以總共輸出了4個 Tasks x 3個分類文件 = 12個本地小文件。
1：shuffle write階段
主要就是在一個stage結束計算之後，爲了下一個stage可以執行shuffle類的算子(比如reduceByKey，groupByKey)，而將每個task處理的數據按key進行「分區」。所謂「分區」，就是對相同的key執行hash算法，從而將相同key都寫入同一個磁盤文件中，而每一個磁盤文件都只屬於reduce端的stage的一個task。在將數據寫入磁盤之前，會先將數據寫入內存緩衝中，當內存緩衝填滿之後，纔會溢寫到磁盤文件中去。
那麼每個執行shuffle write的task，要爲下一個stage創建多少個磁盤文件呢?很簡單，下一個stage的task有多少個，當前stage的每個task就要創建多少份磁盤文件。比如下一個stage總共有100個task，那麼當前stage的每個task都要創建100份磁盤文件。如果當前stage有50個task，總共有10個Executor，每個Executor執行5個Task，那麼每個Executor上總共就要創建500個磁盤文件，所有Executor上會創建5000個磁盤文件。由此可見，未經優化的shuffle write操作所產生的磁盤文件的數量是極其驚人的。
2：shuffle read階段
shuffle read，通常就是一個stage剛開始時要做的事情。此時該stage的每一個task就需要將上一個stage的計算結果中的所有相同key，從各個節點上通過網絡都拉取到自己所在的節點上，然後進行key的聚合或連接等操作。由於shuffle write的過程中，task給Reduce端的stage的每個task都創建了一個磁盤文件，因此shuffle read的過程中，每個task只要從上游stage的所有task所在節點上，拉取屬於自己的那一個磁盤文件即可。
shuffle read的拉取過程是一邊拉取一邊進行聚合的。每個shuffle read task都會有一個自己的buffer緩衝，每次都只能拉取與buffer緩衝相同大小的數據，然後通過內存中的一個Map進行聚合等操作。聚合完一批數據後，再拉取下一批數據，並放到buffer緩衝中進行聚合操作。以此類推，直到最後將所有數據到拉取完，並得到最終的結果。
注意：
1).buffer起到的是緩存作用，緩存能夠加速寫磁盤，提高計算的效率,buffer的默認大小32k。
分區器：根據hash/numRedcue取模決定數據由幾個Reduce處理，也決定了寫入幾個buffer中
block file：磁盤小文件，從圖中我們可以知道磁盤小文件的個數計算公式：
block file=M*R
2).M爲map task的數量，R爲Reduce的數量，一般Reduce的數量等於buffer的數量，都是由分區器決定的
Hash shuffle普通機制的問題
1).Shuffle前在磁盤上會產生海量的小文件，建立通信和拉取數據的次數變多,此時會產生大量耗時低效的 IO 操作 (因為產生過多的小文件)
2).可能導致OOM，大量耗時低效的 IO 操作 ，導致寫磁盤時的對象過多，讀磁盤時候的對象也過多，這些對象存儲在堆內存中，會導致堆內存不足，相應會導致頻繁的GC，GC會導致OOM。由於內存中需要保存海量文件操作句柄和臨時信息，如果數據處理的規模比較龐大的話，內存不可承受，會出現 OOM 等問題。
合併機制的Hash shuffle
合併機制就是複用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個本地小文件。
圖解：
開啓consolidate機制之後，在shuffle write過程中，task就不是爲下游stage的每個task創建一個磁盤文件了。此時會出現shuffleFileGroup的概念，每個shuffleFileGroup會對應一批磁盤文件，磁盤文件的數量與下游stage的task數量是相同的。一個Executor上有多少個CPU core，就可以並行執行多少個task。而第一批並行執行的每個task都會創建一個shuffleFileGroup，並將數據寫入對應的磁盤文件內。
Executor的CPU core執行完一批task，接着執行下一批task時，下一批task就會複用之前已有的shuffleFileGroup，包括其中的磁盤文件。也就是說，此時task會將數據寫入已有的磁盤文件中，而不會寫入新的磁盤文件中。因此，consolidate機制允許不同的task複用同一批磁盤文件，這樣就可以有效將多個task的磁盤文件進行一定程度上的合併，從而大幅度減少磁盤文件的數量，進而提升shuffle write的性能。
假設第二個stage有100個task，第一個stage有50個task，總共還是有10個Executor，每個Executor執行5個task。那麼原本使用未經優化的HashShuffleManager時，每個Executor會產生500個磁盤文件，所有Executor會產生5000個磁盤文件的。但是此時經過優化之後，每個Executor創建的磁盤文件的數量的計算公式爲：CPU core的數量 * 下一個stage的task數量。也就是說，每個Executor此時只會創建100個磁盤文件，所有Executor只會創建1000個磁盤文件。
注意：
1).啓動HashShuffle的合併機制ConsolidatedShuffle的配置:
spark.shuffle.consolidateFiles=true
2).block file=Core*R
Core爲CPU的核數，R爲Reduce的數量
Hash shuffle合併機制的問題
如果 Reducer 端的並行任務或者是數據分片過多的話則 Core * Reducer Task 依舊過大，也會產生很多小文件。
Sort shuffle
SortShuffleManager的運行機制主要分成兩種，一種是普通運行機制，另一種是bypass運行機制。當shuffle read task的數量小於等於spark.shuffle.sort.bypassMergeThreshold參數的值時(默認爲200)，就會啓用bypass機制。
Sort shuffle的普通機制
圖解：
寫入內存數據結構
該圖說明了普通的SortShuffleManager的原理。在該模式下，數據會先寫入一個內存數據結構中(默認5M)，此時根據不同的shuffle算子，可能選用不同的數據結構。如果是reduceByKey這種聚合類的shuffle算子，那麼會選用Map數據結構，一邊通過Map進行聚合，一邊寫入內存;如果是join這種普通的shuffle算子，那麼會選用Array數據結構，直接寫入內存。接着，每寫一條數據進入內存數據結構之後，就會判斷一下，是否達到了某個臨界閾值。如果達到臨界閾值的話，那麼就會嘗試將內存數據結構中的數據溢寫到磁盤，然後清空內存數據結構。
注意：
shuffle中的定時器：定時器會檢查內存數據結構的大小，如果內存數據結構空間不夠，那麼會申請額外的內存，申請的大小滿足如下公式：
applyMemory=nowMenory*2-oldMemory
申請的內存=當前的內存情況*2-上一次的內嵌情況
意思就是說內存數據結構的大小的動態變化，如果存儲的數據超出內存數據結構的大小，將申請內存數據結構存儲的數據*2-內存數據結構的設定值的內存大小空間。申請到了，內存數據結構的大小變大，內存不夠，申請不到，則發生溢寫
排序
在溢寫到磁盤文件之前，會先根據key對內存數據結構中已有的數據進行排序。
溢寫
排序過後，會分批將數據寫入磁盤文件。默認的batch數量是10000條，也就是說，排序好的數據，會以每批1萬條數據的形式分批寫入磁盤文件。寫入磁盤文件是通過Java的BufferedOutputStream實現的。BufferedOutputStream是Java的緩衝輸出流，首先會將數據緩衝在內存中，當內存緩衝滿溢之後再一次寫入磁盤文件中，這樣可以減少磁盤IO次數，提升性能。
merge
一個task將所有數據寫入內存數據結構的過程中，會發生多次磁盤溢寫操作，也就會產生多個臨時文件。最後會將之前所有的臨時磁盤文件都進行合併，這就是merge過程，此時會將之前所有臨時磁盤文件中的數據讀取出來，然後依次寫入最終的磁盤文件之中。此外，由於一個task就只對應一個磁盤文件，也就意味着該task爲Reduce端的stage的task準備的數據都在這一個文件中，因此還會單獨寫一份索引文件，其中標識了下游各個task的數據在文件中的start offset與end offset。
SortShuffleManager由於有一個磁盤文件merge的過程，因此大大減少了文件數量。比如第一個stage有50個task，總共有10個Executor，每個Executor執行5個task，而第二個stage有100個task。由於每個task最終只有一個磁盤文件，因此此時每個Executor上只有5個磁盤文件，所有Executor只有50個磁盤文件。
注意：
1)block file= 2M
一個map task會產生一個索引文件和一個數據大文件
2) m*r>2m(r>2)：SortShuffle會使得磁盤小文件的個數再次的減少
Sort shuffle的bypass機制
bypass運行機制的觸發條件如下：
1)shuffle map task數量小於spark.shuffle.sort.bypassMergeThreshold參數的值。
2)不是聚合類的shuffle算子(比如reduceByKey)。
此時task會爲每個reduce端的task都創建一個臨時磁盤文件，並將數據按key進行hash然後根據key的hash值，將key寫入對應的磁盤文件之中。當然，寫入磁盤文件時也是先寫入內存緩衝，緩衝寫滿之後再溢寫到磁盤文件的。最後，同樣會將所有臨時磁盤文件都合併成一個磁盤文件，並創建一個單獨的索引文件。
該過程的磁盤寫機制其實跟未經優化的HashShuffleManager是一模一樣的，因爲都要創建數量驚人的磁盤文件，只是在最後會做一個磁盤文件的合併而已。因此少量的最終磁盤文件，也讓該機制相對未經優化的HashShuffleManager來說，shuffle read的性能會更好。
而該機制與普通SortShuffleManager運行機制的不同在於：
第一，磁盤寫機制不同;
第二，不會進行排序。也就是說，啓用該機制的最大好處在於，shuffle write過程中，不需要進行數據的排序操作，也就節省掉了這部分的性能開銷。
總結：
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拉取自己的數據時，只要根據索引讀取每個磁盤文件中的部分數據即可。