對於大資料計算框架而言，Shuffle階段的設計優劣是決定效能好壞的關鍵因素之一。本文將介紹目前Spark的shuffle實現，並將之與MapReduce進行簡單對比。本文的介紹順序是：shuffle基本概念，MapReduce Shuffle發展史以及Spark Shuffle發展史。

（1）  shuffle基本概念與常見實現方式

shuffle，是一個運算元，表達的是多對多的依賴關係，在類MapReduce計算框架中，是連線Map階段和Reduce階段的紐帶，即每個Reduce Task從每個Map Task產生數的據中讀取一片資料，極限情況下可能觸發M*R個數據拷貝通道（M是Map Task數目，R是Reduce Task數目）。通常shuffle分為兩部分：Map階段的資料準備和Reduce階段的資料拷貝。首先，Map階段需根據Reduce階段的Task數量決定每個Map Task輸出的資料分片數目，有多種方式存放這些資料分片：

1） 儲存在記憶體中或者磁碟上（Spark和MapReduce都存放在磁碟上）；

2） 每個分片一個檔案（現在Spark採用的方式，若干年前MapReduce採用的方式），或者所有分片放到一個數據檔案中，外加一個索引檔案記錄每個分片在資料檔案中的偏移量（現在MapReduce採用的方式）。

在Map端，不同的資料存放方式各有優缺點和適用場景。一般而言，shuffle在Map端的資料要儲存到磁碟上，以防止容錯觸發重算帶來的龐大開銷（如果儲存到Reduce端記憶體中，一旦Reduce Task掛掉了，所有Map Task需要重算）。但資料在磁碟上存放方式有多種可選方案，在MapReduce前期設計中，採用了現在Spark的方案（目前一直在改進），每個Map Task為每個Reduce Task產生一個檔案，該檔案只儲存特定Reduce Task需處理的資料，這樣會產生M*R個檔案，如果M和R非常龐大，比如均為1000，則會產生100w個檔案，產生和讀取這些檔案會產生大量的隨機IO，效率非常低下。解決這個問題的一種直觀方法是減少檔案數目，常用的方法有：1) 將一個節點上所有Map產生的檔案合併成一個大檔案（MapReduce現在採用的方案），2) 每個節點產生{(slot數目)*R}個檔案（Spark優化後的方案）。對後面這種方案簡單解釋一下：不管是MapReduce 1.0還是Spark，每個節點的資源會被抽象成若干個slot，由於一個Task佔用一個slot，因此slot數目可看成是最多同時執行的Task數目。如果一個Job的Task數目非常多，限於slot數目有限，可能需要執行若干輪。這樣，只需要由第一輪產生{(slot數目)*R}個檔案，後續幾輪產生的資料追加到這些檔案末尾即可。因此，後一種方案可減少大作業產生的檔案數目。

在Reduce端，各個Task會併發啟動多個執行緒同時從多個Map Task端拉取資料。由於Reduce階段的主要任務是對資料進行按組規約。也就是說，需要將資料分成若干組，以便以組為單位進行處理。大家知道，分組的方式非常多，常見的有：Map/HashTable（key相同的，放到同一個value list中）和Sort（按key進行排序，key相同的一組，經排序後會挨在一起），這兩種方式各有優缺點，第一種複雜度低，效率高，但是需要將資料全部放到記憶體中，第二種方案複雜度高，但能夠藉助磁碟（外部排序）處理龐大的資料集。Spark前期採用了第一種方案，而在最新的版本中加入了第二種方案， MapReduce則從一開始就選用了基於sort的方案。

（2） MapReduce Shuffle發展史

【階段1】：MapReduce Shuffle的發展也並不是一馬平川的，剛開始（0.10.0版本之前）採用了“每個Map Task產生R個檔案”的方案，前面提到，該方案會產生大量的隨機讀寫IO，對於大資料處理而言，非常不利。

【階段2】：為了避免Map Task產生大量檔案，HADOOP-331嘗試對該方案進行優化，優化方法：為每個Map Task提供一個環形buffer，一旦buffer滿了後，則將記憶體資料spill到磁碟上（外加一個索引檔案，儲存每個partition的偏移量），最終合併產生的這些spill檔案，同時建立一個索引檔案，儲存每個partition的偏移量。

（階段2）：這個階段並沒有對shuffle架構做調成，只是對shuffle的環形buffer進行了優化。在Hadoop 2.0版本之前，對MapReduce作業進行引數調優時，Map階段的buffer調優非常複雜的，涉及到多個引數，這是由於buffer被切分成兩部分使用：一部分儲存索引（比如parition、key和value偏移量和長度），一部分儲存實際的資料，這兩段buffer均會影響spill檔案數目，因此，需要根據資料特點對多個引數進行調優，非常繁瑣。而MAPREDUCE-64則解決了該問題，該方案讓索引和資料共享一個環形緩衝區，不再將其分成兩部分獨立使用，這樣只需設定一個引數控制spill頻率。

【階段3（進行中）】：目前shuffle被當做一個子階段被嵌到Reduce階段中的。由於MapReduce模型中，Map Task和Reduce Task可以同時執行，因此一個作業前期啟動的Reduce Task將一直處於shuffle階段，直到所有Map Task執行完成，而在這個過程中，Reduce Task佔用著資源，但這部分資源利用率非常低，基本上只使用了IO資源。為了提高資源利用率，一種非常好的方法是將shuffle從Reduce階段中獨立處理，變成一個獨立的階段/服務，由專門的shuffler service負責資料拷貝，目前百度已經實現了該功能（準備開源？），且收益明顯，具體參考：MAPREDUCE-2354。

（3） Spark Shuffle發展史

目前看來，Spark Shuffle的發展史與MapReduce發展史非常類似。初期Spark在Map階段採用了“每個Map Task產生R個檔案”的方法，在Reduce階段採用了map分組方法，但隨Spark變得流行，使用者逐漸發現這種方案在處理大資料時存在嚴重瓶頸問題，因此嘗試對Spark進行優化和改進，相關連結有：External Sorting for Aggregator and CoGroupedRDDs，“Optimizing Shuffle Performance in Spark”，“Consolidating Shuffle Files in Spark”，優化動機和思路與MapReduce非常類似。

Spark在前期設計中過多依賴於記憶體，使得一些執行在MapReduce之上的大作業難以直接執行在Spark之上（可能遇到OOM問題）。目前Spark在處理大資料集方面尚不完善，使用者需根據作業特點選擇性的將一部分作業遷移到Spark上，而不是整體遷移。隨著Spark的完善，很多內部關鍵模組的設計思路將變得與MapReduce升級版Tez非常類似。