轉載自：https://www.cnblogs.com/itboys/p/9226479.html
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拉取自己的數據時，只要根據索引讀取每個磁盤文件中的部分數據即可。