Spark Shuffle和Hadoop Shuffle有哪些區別

本篇內容主要講解“Spark Shuffle和Hadoop Shuffle有哪些區別”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Spark Shuffle和Hadoop Shuffle有哪些區別”吧!

Shuffle的本意是洗牌、混洗的意思，把一組有規則的數據盡量打亂成無規則的數據。而在MapReduce中，Shuffle更像是洗牌的逆過程，指的是將map端的無規則輸出按指定的規則“打亂”成具有一定規則的數據，以便reduce端接收處理。其在MapReduce中所處的工作階段是map輸出后到reduce接收前，具體可以分為map端和reduce端前后兩個部分。

在Shuffle之前，也就是在map階段，MapReduce會對要處理的數據進行分片（split）操作，為每一個分片分配一個MapTask任務。接下來map會對每一個分片中的每一行數據進行處理得到鍵值對（key,value）此時得到的鍵值對又叫做“中間結果”。此后便進入reduce階段，由此可以看出Shuffle階段的作用是處理“中間結果”。

由于Shuffle涉及到了磁盤的讀寫和網絡的傳輸，因此Shuffle性能的高低直接影響到了整個程序的運行效率。

我們先放一張圖，下圖說明了二者shuffle的主要區別：

MapReduce Shuffle

我們在《大數據嗶嗶集20210107》中詳細講解過MapReduce的shuffle過程：

map階段

1. 在map task執行時，它的輸入數據來源于HDFS的block，當然在MapReduce概念中，map task只讀取split。Split與block的對應關系可能是多對一，默認是一對一。

2. 在經過mapper的運行后，我們得知mapper的輸出是這樣一個key/value對： key是“hello”， value是數值1。因為當前map端只做加1的操作，在reduce task里才去合并結果集。這個job有3個reduce task，到底當前的“hello”應該交由哪個reduce去做呢，是需要現在決定的。

• 分區（partition） MapReduce提供Partitioner接口，它的作用就是根據key或value及reduce的數量來決定當前的這對輸出數據最終應該交由哪個reduce task處理。默認對key hash后再以reduce task數量取模。默認的取模方式只是為了平均reduce的處理能力，如果用戶自己對Partitioner有需求，可以訂制并設置到job上。 一個split被分成了3個partition。

• 排序sort 在spill寫入之前，會先進行二次排序，首先根據數據所屬的partition進行排序，然后每個partition中的數據再按key來排序。partition的目是將記錄劃分到不同的Reducer上去，以期望能夠達到負載均衡，以后的Reducer就會根據partition來讀取自己對應的數據。接著運行combiner(如果設置了的話)，combiner的本質也是一個Reducer，其目的是對將要寫入到磁盤上的文件先進行一次處理，這樣，寫入到磁盤的數據量就會減少。

• 溢寫（spill） Map端會處理輸入數據并產生中間結果，這個中間結果會寫到本地磁盤，而不是HDFS。每個Map的輸出會先寫到內存緩沖區中， 緩沖區的作用是批量收集map結果，減少磁盤IO的影響。我們的key/value對以及Partition的結果都會被寫入緩沖區。當然寫入之前，key與value值都會被序列化成字節數組。 當寫入的數據達到設定的閾值時，系統將會啟動一個線程將緩沖區的數據寫到磁盤，這個過程叫做spill。 這個溢寫是由單獨線程來完成，不影響往緩沖區寫map結果的線程。溢寫線程啟動時不應該阻止map的結果輸出，所以整個緩沖區有個溢寫的比例spill.percent。這個比例默認是0.8。 將數據寫到本地磁盤產生spill文件(spill文件保存在{mapred.local.dir}指定的目錄中，MapReduce任務結束后就會被刪除)。

• 合并（merge） 每個Map任務可能產生多個spill文件，在每個Map任務完成前，會通過多路歸并算法將這些spill文件歸并成一個文件。這個操作就叫merge(spill文件保存在{mapred.local.dir}指定的目錄中，Map任務結束后就會被刪除)。一個map最終會溢寫一個文件。 至此，Map的shuffle過程就結束了。

Reduce階段

Reduce端的shuffle主要包括三個階段，copy、sort(merge)和reduce。

• copy 首先要將Map端產生的輸出文件拷貝到Reduce端，但每個Reducer如何知道自己應該處理哪些數據呢？因為Map端進行partition的時候，實際上就相當于指定了每個Reducer要處理的數據(partition就對應了Reducer)，**所以Reducer在拷貝數據的時候只需拷貝與自己對應的partition中的數據即可。**每個Reducer會處理一個或者多個partition，但需要先將自己對應的partition中的數據從每個Map的輸出結果中拷貝過來。

• merge Copy過來的數據會先放入內存緩沖區中，這里的緩沖區大小要比map端的更為靈活，它基于JVM的heap size設置，因為Shuffle階段Reducer不運行，所以應該把絕大部分的內存都給Shuffle用。

這里需要強調的是：

merge階段，也稱為sort階段，因為這個階段的主要工作是執行了歸并排序。從Map端拷貝到Reduce端的數據都是有序的，所以很適合歸并排序。

merge有三種形式：1)內存到內存 2)內存到磁盤 3)磁盤到磁盤。默認情況下第一種形式不啟用，讓人比較困惑，是吧。

當copy到內存中的數據量到達一定閾值，就啟動內存到磁盤的merge，即第二種merge方式，與map 端類似，這也是溢寫的過程，這個過程中如果你設置有Combiner，也是會啟用的，然后在磁盤中生成了眾多的溢寫文件。這種merge方式一直在運行，直到沒有map端的數據時才結束。然后啟動第三種磁盤到磁盤的merge方式生成最終的那個文件。

• reduce 不斷地merge后，最后會生成一個“最終文件”。為什么加引號？因為這個文件可能存在于磁盤上，也可能存在于內存中。對我們來說，當然希望它存放于內存中，直接作為Reducer的輸入，但默認情況下，這個文件是存放于磁盤中的。至于怎樣才能讓這個文件出現在內存中，參見性能優化篇。 然后就是Reducer執行，在這個過程中產生了最終的輸出結果，并將其寫到HDFS上。

Spark Shuffle

我們在之前的文章《Spark性能優化總結》中提到過，Spark Shuffle 的原理和演進過程。

Spark在DAG階段以寬依賴shuffle為界，劃分stage，上游stage做map task，每個map task將計算結果數據分成多份，每一份對應到下游stage的每個partition中，并將其臨時寫到磁盤，該過程叫做shuffle write。

下游stage做reduce task，每個reduce task通過網絡拉取上游stage中所有map task的指定分區結果數據，該過程叫做shuffle read，最后完成reduce的業務邏輯。

下圖中，上游stage有3個map task，下游stage有4個reduce task，那么這3個map task中每個map task都會產生4份數據。而4個reduce task中的每個reduce task都會拉取上游3個map task對應的那份數據。

Spark Shuffle演進

• < 0.8 hashBasedShuffle 每個map端的task為每個reduce端的partition/task生成一個文件，通常會產生大量的文件，伴隨大量的隨機磁盤IO操作與大量的內存開銷M * R

• 0.8.1 引入文件合并File Consolidation機制 每個executor為每個reduce端的partition生成一個文件E*R

• 0.9 引入External AppendOnlyMap combine時可以將數據spill到磁盤，然后通過堆排序merge

• 1.1 引入sortBasedShuffle 每個map task不會為每個reducer task生成一個單獨的文件，而是會將所有的結果寫到一個文件里，同時會生成一個index文件，reducer可以通過這個index文件取得它需要處理的數據M

• 1.4 引入Tungsten-Sort Based Shuffle 亦稱unsafeShuffle，將數據記錄用序列化的二進制方式存儲，把排序轉化成指針數組的排序，引入堆外內存空間和新的內存管理模型

• 1.6 Tungsten-sort并入Sort Based Shuffle 由SortShuffleManager自動判斷選擇最佳Shuffle方式，如果檢測到滿足Tungsten-sort條件會自動采用Tungsten-sort Based Shuffle，否則采用Sort Based Shuffle

• 2.0 hashBasedShuffle退出歷史舞臺 從此Spark只有sortBasedShuffle

總結

到此為止，我們已經把二者的原理完完整整的講了一遍。最后，總結引用ITStar總結過的二者的不同精簡要點版本：

Hadoop Shuffle:通過Map端處理的數據到Reduce端的中間的過程就是Shuffle.

Spark Shuffle:在DAG調度過程中,stage階段的劃分是根據shuffle過程,也就是存在ShuffleDependency寬窄依賴的時候,需要進行shuffle,(這時候會將作業Job劃分成多個stage;并且在劃分stage的時候,構建shuffleDependency的時候進行shuffle注冊,獲取后續數據讀取所需要的shuffleHandle),最終每一個Job提交后都會生成一個ResultStage和若干個ShuffleMapStage.

shuffle過程排序次數不同

Hadoop Shuffle過程中總共發生3次排序,詳細分別如下:

第一次排序行為:在map階段,由環形緩沖區溢出到磁盤上時,落地磁盤的文件會按照key進行分區和排序,屬于分區內有序,排序算法為快速排序.

第二次排序行為:在map階段,對溢出的文件進行combiner合并過程中,需要對溢出的小文件進行歸檔排序,合并,排序算法為歸并排序.

第三次排序行為:在map階段,reduce task將不同map task端文件拉取到同一個reduce分區后,對文件進行合并,排序,排序算法為歸并排序.

spark shuffle過程在滿足shuffle manager為sortshuffleManager,且運行模式為普通模式的情況下才會發生排序行為,排序行為發生在數據結構中保存數據內存達到閥值,再溢出磁盤文件之前會對內存數據結構中數據進行排序;

spark中sorted-Based Shuffle在Mapper端是進行排序的,包括partition的排序和每個partition內部元素進行排序,但是在Reducer端沒有進行排序,所有job的結果默認情況下不是排序的.Sprted-Based Shuffle 采用 Tim-Sort排序算法,好處是可以極為高效的使用Mapper端的排序成果完成全局排序.

shuffle 邏輯流劃分

Hadoop是基于文件的數據結構，Spark是基于RDD的數據結構,計算性能要比Hadoop要高。

Shuffle Fetch后數據存放位置

Hadoopreduce 端將 map task 的文件拉取到同一個reduce分區,是將文件進行歸并排序,合并,將文件直接保存在磁盤上。

SparkShuffle Read 拉取來的數據首先肯定是放在Reducer端的內存緩存區中的,現在的實現都是內存+磁盤的方式(數據結構使用 ExternalAppendOnlyMap),當然也可以通過Spark.shuffle.spill=false來設置只能使用內存.使用ExternalAppendOnlyMap的方式時候如果內存的使用達到一定臨界值,會首先嘗試在內存中擴大ExternalAppendOnlyMap(內部有實現算法),如果不能擴容的話才會spil到磁盤.

什么時候進行Shuffle Fetch操作

Hadoop Shuffle把數據拉過來之后,然后進行計算，如果用MapReduce求平均值的話,它的算法就會很好實現。Spark Shuffle的過程是邊拉取數據邊進行Aggregrate操作。

Fetch操作與數據計算粒度

Hadoop的MapReduce是粗粒度的，Hadoop Shuffle Reducer Fetch 到的數據record先暫時被存放到Buffer中,當Buffer快滿時才進行combine()操作。Spark的Shuffle Fetch是細粒度的,Reducer是對Map端數據Record邊拉取邊聚合。

性能優化的角度

Hadoop MapReduce的shuffle方式單一.Spark針對不同類型的操作，不同類型的參數,會使用不同的shuffle write方式;而spark更加全面。

到此，相信大家對“Spark Shuffle和Hadoop Shuffle有哪些區別”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！