面試常見問題，必備答案。

參考：https://blog.csdn.net/u010697988/article/details/70173104

mapReducehe和Spark之間的最大區別是前者較偏向於離線處理，而後者重視實效性，下面主要介紹mapReducehe和Spark兩者的shuffle過程。

MapReduce的Shuffle過程

MapReduce計算模型一般包括兩個重要的階段：Map是映射，負責數據的過濾分發；Reduce是規約，負責數據的計算歸並。Reduce的數據來源於Map，Map的輸出即是Reduce的輸入，Reduce需要通過 Shuffle來獲取數據。

Spill過程

Spill過程包括輸出、排序、溢寫、合並等步驟，如圖所示：

Collect（此過程較難理解，可簡化說明）

每個Map任務不斷地以對的形式把數據輸出到在內存中構造的一個環形數據結構中。使用環形數據結構是為了更有效地使用內存空間，在內存中放置盡可能多的數據。

這個數據結構其實就是個字節數組，叫Kvbuffer， 名如其義，但是這裏面不光放置了數據，還放置了一些索引數據，給放置索引數據的區域起了一個Kvmeta的別名，在Kvbuffer的一塊區域上穿了一個 IntBuffer（字節序采用的是平臺自身的字節序）的馬甲。數據區域和索引數據區域在Kvbuffer中是相鄰不重疊的兩個區域，用一個分界點來劃分兩者，分界點不是亙古不變的，而是每次 Spill之後都會更新一次。初始的分界點是0，數據的存儲方向是向上增長，索引數據的存儲方向是向下增長，如圖所示：

Kvbuffer的存放指針bufindex是一直悶著頭地向上增長，比如bufindex初始值為0，一個Int型的key寫完之後，bufindex增長為4，一個Int型的value寫完之後，bufindex增長為8。

索引是對在kvbuffer中的索引，是個四元組，包括：value的起始位置、key的起始位置、partition值、value的長度， 占用四個Int長度，Kvmeta的存放指針Kvindex每次都是向下跳四個“格子”，然後再向上一個格子一個格子地填充四元組的數據。比如 Kvindex初始位置是-4，當第一個寫完之後，(Kvindex+0)的位置存放value的起始位置、(Kvindex+1)的位置存放key的起始位置、 (Kvindex+2)的位置存放partition的值、(Kvindex+3)的位置存放value的長度，然後Kvindex跳到-8位置，等第二 個和索引寫完之後，Kvindex跳到-32位置。

Kvbuffer的大小雖然可以通過參數設置，但是總共就那麽大，和索引不斷地增加，加著加著，Kvbuffer總有不夠用的那天，那怎麽辦？把數據從內存刷到磁盤上再接著往內存寫數據，把 Kvbuffer中的數據刷到磁盤上的過程就叫Spill，多麽明了的叫法，內存中的數據滿了就自動地spill到具有更大空間的磁盤。

關於Spill 觸發的條件，也就是Kvbuffer用到什麽程度開始Spill，還是要講究一下的。如果把Kvbuffer用得死死得，一點縫都不剩的時候再開始 Spill，那Map任務就需要等Spill完成騰出空間之後才能繼續寫數據；如果Kvbuffer只是滿到一定程度，比如80%的時候就開始 Spill，那在Spill的同時，Map任務還能繼續寫數據，如果Spill夠快，Map可能都不需要為空閑空間而發愁。兩利相衡取其大，一般選擇後 者。

Spill這個重要的過程是由Spill線程承擔，Spill線程從Map任務接到“命令”之後就開始正式幹活，幹的活叫SortAndSpill，原來不僅僅是Spill，在Spill之前還有個頗具爭議性的Sort。

Sort

先把Kvbuffer中的數據按照partition值和key兩個關鍵字升序排序，移動的只是索引數據，排序結果是Kvmeta中數據按照partition為單位聚集在一起，同一partition內的按照key有序。

Spill

Spill線程為這次Spill過程創建一個磁盤文件：從所有的本地目錄中輪訓查找能存儲這麽大空間的目錄，找到之後在其中創建一個類似於 “spill12.out”的文件。Spill線程根據排過序的Kvmeta挨個partition的把數據吐到這個文件中，一個partition對應的數據吐完之後順序地吐下個partition，直到把所有的partition遍歷 完。一個partition在文件中對應的數據也叫段(segment)。

所有的partition 對應的數據都放在這個文件裏，雖然是順序存放的，但是怎麽直接知道某個partition在這個文件中存放的起始位置呢？強大的索引又出場了。有一個三元 組記錄某個partition對應的數據在這個文件中的索引：起始位置、原始數據長度、壓縮之後的數據長度，一個partition對應一個三元組。然後 把這些索引信息存放在內存中，如果內存中放不下了，後續的索引信息就需要寫到磁盤文件中了：從所有的本地目錄中輪訓查找能存儲這麽大空間的目錄，找到之後 在其中創建一個類似於“spill12.out.index”的文件，文件中不光存儲了索引數據，還存儲了crc32的校驗數據。 (spill12.index不一定在磁盤上創建，如果內存（默認1M空間）中能放得下就放在內存中，即使在磁盤上創建了，和 spill12.out文件也不一定在同一個目錄下。)

每一次Spill過程就會最少生成一個out文件，有時還會生成index文件，Spill的次數也烙印在文件名中。索引文件和數據文件的對應關系如下圖所示：

話 分兩端，在Spill線程如火如荼的進行SortAndSpill工作的同時，Map任務不會因此而停歇，而是一無既往地進行著數據輸出。Map還是把數 據寫到kvbuffer中，那問題就來了：只顧著悶頭按照bufindex指針向上增長，kvmeta只顧著按照Kvindex向下增長，是保持指針起始位置不變繼續跑呢，還是 另謀它路？如果保持指針起始位置不變，很快bufindex和Kvindex就碰頭了，碰頭之後再重新開始或者移動內存都比較麻煩，不可取。Map取 kvbuffer中剩余空間的中間位置，用這個位置設置為新的分界點，bufindex指針移動到這個分界點，Kvindex移動到這個分界點的-16位 置，然後兩者就可以和諧地按照自己既定的軌跡放置數據了，當Spill完成，空間騰出之後，不需要做任何改動繼續前進。分界點的轉換如下圖所示：（此部分為collect過程，可簡化）

Map任務總要把輸出的數據寫到磁盤上，即使輸出數據量很小在內存中全部能裝得下，在最後也會把數據刷到磁盤上。

Merge

Map任務如果輸出數據量很大，可能會進行好幾次Spill，out文件和Index文件會產生很多，分布在不同的磁盤上。最後把這些文件進行合並的merge過程閃亮登場。

Merge 過程怎麽知道產生的Spill文件都在哪了呢？從所有的本地目錄上掃描得到產生的Spill文件，然後把路徑存儲在一個數組裏。Merge過程又怎麽知道 Spill的索引信息呢？沒錯，也是從所有的本地目錄上掃描得到Index文件，然後把索引信息存儲在一個列表裏。到這裏，又遇到了一個值得納悶的地方。 在之前Spill過程中的時候為什麽不直接把這些信息存儲在內存中呢，何必又多了這步掃描的操作？特別是Spill的索引數據，之前當內存超限之後就把數 據寫到磁盤，現在又要從磁盤把這些數據讀出來，還是需要裝到更多的內存中。之所以多此一舉，是因為這時kvbuffer這個內存大戶已經不再使用可以回 收，有內存空間來裝這些數據了。（對於內存空間較大的土豪來說，用內存來省卻這兩個io步驟還是值得考慮的。）

然後為merge過程創建一個叫file.out的文件和一個叫file.out.Index的文件用來存儲最終的輸出和索引。

一 個partition一個partition的進行合並輸出。對於某個partition來說，從索引列表中查詢這個partition對應的所有索引信 息，每個對應一個段插入到段列表中。也就是這個partition對應一個段列表，記錄所有的Spill文件中對應的這個partition那段數據的文 件名、起始位置、長度等等。

然後對這個partition對應的所有的segment進行合並，目標是合並成一個segment。當這個 partition對應很多個segment時，會分批地進行合並：先從segment列表中把第一批取出來，以key為關鍵字放置成最小堆，然後從最小 堆中每次取出最小的輸出到一個臨時文件中，這樣就把這一批段合並成一個臨時的段，把它加回到segment列表中；再從segment列表中把第二批取出 來合並輸出到一個臨時segment，把其加入到列表中；這樣往復執行，直到剩下的段是一批，輸出到最終的文件中。

最終的索引數據仍然輸出到Index文件中。

Map端的Shuffle過程到此結束。

Copy

Reduce 任務通過HTTP向各個Map任務拖取它所需要的數據。每個節點都會啟動一個常駐的HTTP server，其中一項服務就是響應Reduce拖取Map數據。當有MapOutput的HTTP請求過來的時候，HTTP server就讀取相應的Map輸出文件中對應這個Reduce部分的數據通過網絡流輸出給Reduce。

Reduce任務拖取某個Map 對應的數據，如果在內存中能放得下這次數據的話就直接把數據寫到內存中。Reduce要向每個Map去拖取數據，在內存中每個Map對應一塊數據，當內存 中存儲的Map數據占用空間達到一定程度的時候，開始啟動內存中merge，把內存中的數據merge輸出到磁盤上一個文件中。

如果在內存 中不能放得下這個Map的數據的話，直接把Map數據寫到磁盤上，在本地目錄創建一個文件，從HTTP流中讀取數據然後寫到磁盤，使用的緩存區大小是 64K。拖一個Map數據過來就會創建一個文件，當文件數量達到一定閾值時，開始啟動磁盤文件merge，把這些文件合並輸出到一個文件。

有些Map的數據較小是可以放在內存中的，有些Map的數據較大需要放在磁盤上，這樣最後Reduce任務拖過來的數據有些放在內存中了有些放在磁盤上，最後會對這些來一個全局合並。

Merge Sort

這裏使用的Merge和Map端使用的Merge過程一樣。Map的輸出數據已經是有序的，Merge進行一次合並排序，所謂Reduce端的 sort過程就是這個合並的過程。一般Reduce是一邊copy一邊sort，即copy和sort兩個階段是重疊而不是完全分開的。

Reduce端的Shuffle過程至此結束。

Spark的Shuffle過程

Shuffle Writer

Spark豐富了任務類型，有些任務之間數據流轉不需要通過Shuffle，但是有些任務之間還是需要通過Shuffle來傳遞數據，比如wide dependency的group by key。

Spark中需要Shuffle 輸出的Map任務會為每個Reduce創建對應的bucket，Map產生的結果會根據設置的partitioner得到對應的bucketId，然後填 充到相應的bucket中去。每個Map的輸出結果可能包含所有的Reduce所需要的數據，所以每個Map會創建R個bucket（R是reduce的 個數），M個Map總共會創建M*R個bucket。

Map創建的bucket其實對應磁盤上的一個文 件，Map的結果寫到每個bucket中其實就是寫到那個磁盤文件中，這個文件也被稱為blockFile，是Disk Block Manager管理器通過文件名的Hash值對應到本地目錄的子目錄中創建的。每個Map要在節點上創建R個磁盤文件用於結果輸出，Map的結果是直接輸 出到磁盤文件上的，100KB的內存緩沖是用來創建Fast Buffered OutputStream輸出流。這種方式一個問題就是Shuffle文件過多。

針對上述Shuffle 過程產生的文件過多問題，Spark有另外一種改進的Shuffle過程：consolidation Shuffle，以期顯著減少Shuffle文件的數量。在consolidation Shuffle中每個bucket並非對應一個文件，而是對應文件中的一個segment部分。Job的map在某個節點上第一次執行，為每個 reduce創建bucket對應的輸出文件，把這些文件組織成ShuffleFileGroup，當這次map執行完之後，這個 ShuffleFileGroup可以釋放為下次循環利用；當又有map在這個節點上執行時，不需要創建新的bucket文件，而是在上次的 ShuffleFileGroup中取得已經創建的文件繼續追加寫一個segment；當前次map還沒執行完，ShuffleFileGroup還沒有 釋放，這時如果有新的map在這個節點上執行，無法循環利用這個ShuffleFileGroup，而是只能創建新的bucket文件組成新的 ShuffleFileGroup來寫輸出。

比 如一個Job有3個Map和2個reduce：(1) 如果此時集群有3個節點有空槽，每個節點空閑了一個core，則3個Map會調度到這3個節點上執行，每個Map都會創建2個Shuffle文件，總共創 建6個Shuffle文件；(2) 如果此時集群有2個節點有空槽，每個節點空閑了一個core，則2個Map先調度到這2個節點上執行，每個Map都會創建2個Shuffle文件，然後其 中一個節點執行完Map之後又調度執行另一個Map，則這個Map不會創建新的Shuffle文件，而是把結果輸出追加到之前Map創建的Shuffle 文件中；總共創建4個Shuffle文件；(3) 如果此時集群有2個節點有空槽，一個節點有2個空core一個節點有1個空core，則一個節點調度2個Map一個節點調度1個Map，調度2個Map的 節點上，一個Map創建了Shuffle文件，後面的Map還是會創建新的Shuffle文件，因為上一個Map還正在寫，它創建的 ShuffleFileGroup還沒有釋放；總共創建6個Shuffle文件。

Shuffle Fetcher

Reduce去拖Map的輸出數據，Spark提供了兩套不同的拉取數據框架：通過socket連接去取數據；使用netty框架去取數據。

每 個節點的Executor會創建一個BlockManager，其中會創建一個BlockManagerWorker用於響應請求。當Reduce的 GET_BLOCK的請求過來時，讀取本地文件將這個blockId的數據返回給Reduce。如果使用的是Netty框架，BlockManager會 創建ShuffleSender用於發送Shuffle數據。

並不是所有的數據都是通過網絡讀取，對於在本節點的Map數據，Reduce直接去磁盤上讀取而不再通過網絡框架。

Reduce 拖過來數據之後以什麽方式存儲呢？Spark Map輸出的數據沒有經過排序，Spark Shuffle過來的數據也不會進行排序，Spark認為Shuffle過程中的排序不是必須的，並不是所有類型的Reduce需要的數據都需要排序，強 制地進行排序只會增加Shuffle的負擔。Reduce拖過來的數據會放在一個HashMap中，HashMap中存儲的也是對，key是Map輸出的key，Map輸出對應這個key的所有value組成HashMap的value。Spark將 Shuffle取過來的每一個對插入或者更新到HashMap中，來一個處理一個。HashMap全部放在內存中。

Shuffle 取過來的數據全部存放在內存中，對於數據量比較小或者已經在Map端做過合並處理的Shuffle數據，占用內存空間不會太大，但是對於比如group by key這樣的操作，Reduce需要得到key對應的所有value，並將這些value組一個數組放在內存中，這樣當數據量較大時，就需要較多內存。

當 內存不夠時，要不就失敗，要不就用老辦法把內存中的數據移到磁盤上放著。Spark意識到在處理數據規模遠遠大於內存空間時所帶來的不足，引入了一個具有 外部排序的方案。Shuffle過來的數據先放在內存中，當內存中存儲的對超過1000並且內存使用超過70%時，判斷節點上可用內存如果還足夠，則把內存緩沖區大小翻倍，如果可用內存不再夠了，則把內存中 的對排序然後寫到磁盤文件中。最後把內存緩沖區中的數據排序之後和那些磁盤文件組成一個最小堆，每次從最小堆中讀取最小的數據，這個和 MapReduce中的merge過程類似。

MapReduce和Spark的Shuffle過程對比

MapReduce Shuffle可優化方向

• 壓縮：對數據進行壓縮，減少寫讀數據量；

• 減少不必要的排序：並不是所有類型的Reduce需要的數據都是需要排序的，排序這個nb的過程如果不需要最好還是不要的好；

• 內存化：Shuffle的數據不放在磁盤而是盡量放在內存中，除非逼不得已往磁盤上放；當然了如果有性能和內存相當的第三方存儲系統，那放在第三方存儲系統上也是很好的；這個是個大招；

• 網絡框架：netty的性能據說要占優了；

• 本節點上的數據不走網絡框架：對於本節點上的Map輸出，Reduce直接去讀吧，不需要繞道網絡框架。

Spark Shuffle可優化方向

Spark作為MapReduce的進階架構，對於Shuffle過程已經是優化了的，特別是對於那些具有爭議的步驟已經做了優化，但是Spark的Shuffle對於我們來說在一些方面還是需要優化的。

• 壓縮：對數據進行壓縮，減少寫讀數據量；

• 內存化：Spark歷史版本中是有這樣設計的：Map寫數據先把數據全部寫到內存中，寫完之後再把數據刷到磁盤上；考慮內存是緊缺資源，後來修改成把數據直接寫到磁盤了；對於具有較大內存的集群來講，還是盡量地往內存上寫吧，內存放不下了再放磁盤。

MapReduce和spark的shuffle過程詳解