共計 5298 個字符，預計需要花費 14 分鐘才能閱讀完成。
這篇文章主要介紹“MapReduce Shuffle 過程是怎樣的”，在日常操作中，相信很多人在 MapReduce Shuffle 過程是怎樣的問題上存在疑惑，丸趣 TV 小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”MapReduce Shuffle 過程是怎樣的”的疑惑有所幫助！接下來，請跟著丸趣 TV 小編一起來學習吧！
Shuffle 的正常意思是洗牌或弄亂，可能大家更熟悉的是 Java API 里的 Collections.shuffle(List) 方法，它會隨機地打亂參數(shù) list 里的元素順序。如果你不知道 MapReduce 里 Shuffle 是什么，那么請看這張圖：
這張是官方對 Shuffle 過程的描述。但我可以肯定的是，單從這張圖你基本不可能明白 Shuffle 的過程，因為它與事實相差挺多，細節(jié)也是錯亂的。后面我會具體描述 Shuffle 的事實情況，所以這里你只要清楚 Shuffle 的大致范圍就成-怎樣把 map task 的輸出結(jié)果有效地傳送到 reduce 端。也可以這樣理解，Shuffle 描述著數(shù)據(jù)從 map task 輸出到 reduce task 輸入的這段過程。
在 Hadoop 這樣的集群環(huán)境中，大部分 map task 與 reduce task 的執(zhí)行是在不同的節(jié)點上。當然很多情況下 Reduce 執(zhí)行時需要跨節(jié)點去拉取其它節(jié)點上的 map task 結(jié)果。如果集群正在運行的 job 有很多，那么 task 的正常執(zhí)行對集群內(nèi)部的網(wǎng)絡資源消耗會很嚴重。這種網(wǎng)絡消耗是正常的，我們不能限制，能做的就是最大化地減少不必要的消耗。還有在節(jié)點內(nèi)，相比于內(nèi)存，磁盤 IO 對 job 完成時間的影響也是可觀的。從最基本的要求來說，我們對 Shuffle 過程的期望可以有：完整地從 map task 端拉取數(shù)據(jù)到 reduce 端。在跨節(jié)點拉取數(shù)據(jù)時，盡可能地減少對帶寬的不必要消耗。減少磁盤 IO 對 task 執(zhí)行的影響。
OK，看到這里時，大家可以先停下來想想，如果是自己來設計這段 Shuffle 過程，那么你的設計目標是什么。我想能優(yōu)化的地方主要在于減少拉取數(shù)據(jù)的量及盡量使用內(nèi)存而不是磁盤。
我的分析是基于 Hadoop0.21.0 的源碼，如果與你所認識的 Shuffle 過程有差別，不吝指出。我會以 WordCount 為例，并假設它有 8 個 map task 和 3 個 reduce task。從上圖看出，Shuffle 過程橫跨 map 與 reduce 兩端，所以下面我也會分兩部分來展開。
先看看 map 端的情況，如下圖：
上圖可能是某個 map task 的運行情況。拿它與官方圖的左半邊比較，會發(fā)現(xiàn)很多不一致。官方圖沒有清楚地說明 partition，sort 與 combiner 到底作用在哪個階段。我畫了這張圖，希望讓大家清晰地了解從 map 數(shù)據(jù)輸入到 map 端所有數(shù)據(jù)準備好的全過程。
整個流程我分了四步。簡單些可以這樣說，每個 map task 都有一個內(nèi)存緩沖區(qū)，存儲著 map 的輸出結(jié)果，當緩沖區(qū)快滿的時候需要將緩沖區(qū)的數(shù)據(jù)以一個臨時文件的方式存放到磁盤，當整個 map task 結(jié)束后再對磁盤中這個 map task 產(chǎn)生的所有臨時文件做合并，生成最終的正式輸出文件，然后等待 reduce task 來拉數(shù)據(jù)。
當然這里的每一步都可能包含著多個步驟與細節(jié)，下面我對細節(jié)來一一說明：
在 map task 執(zhí)行時，它的輸入數(shù)據(jù)來源于 HDFS 的 block，當然在 MapReduce 概念中，map task 只讀取 split。Split 與 block 的對應關系可能是多對一，默認是一對一。在 WordCount 例子里，假設 map 的輸入數(shù)據(jù)都是像“aaa”這樣的字符串。 在經(jīng)過 mapper 的運行后，我們得知 mapper 的輸出是這樣一個 key/value 對： key 是“aaa”， value 是數(shù)值 1。因為當前 map 端只做加 1 的操作，在 reduce task 里才去合并結(jié)果集。前面我們知道這個 job 有 3 個 reduce task，到底當前的“aaa”應該交由哪個 reduce 去做呢，是需要現(xiàn)在決定的。
MapReduce 提供 Partitioner 接口，它的作用就是根據(jù) key 或 value 及 reduce 的數(shù)量來決定當前的這對輸出數(shù)據(jù)最終應該交由哪個 reduce task 處理。默認對 key hash 后再以 reduce task 數(shù)量取模。默認的取模方式只是為了平均 reduce 的處理能力，如果用戶自己對 Partitioner 有需求，可以訂制并設置到 job 上。
在我們的例子中，“aaa”經(jīng)過 Partitioner 后返回 0，也就是這對值應當交由第一個 reducer 來處理。接下來，需要將數(shù)據(jù)寫入內(nèi)存緩沖區(qū)中，緩沖區(qū)的作用是批量收集 map 結(jié)果，減少磁盤 IO 的影響。我們的 key/value 對以及 Partition 的結(jié)果都會被寫入緩沖區(qū)。當然寫入之前，key 與 value 值都會被序列化成字節(jié)數(shù)組。
整個內(nèi)存緩沖區(qū)就是一個字節(jié)數(shù)組，它的字節(jié)索引及 key/value 存儲結(jié)構(gòu)我沒有研究過。如果有朋友對它有研究，那么請大致描述下它的細節(jié)吧。 這個內(nèi)存緩沖區(qū)是有大小限制的，默認是 100MB。當 map task 的輸出結(jié)果很多時，就可能會撐爆內(nèi)存，所以需要在一定條件下將緩沖區(qū)中的數(shù)據(jù)臨時寫入磁盤，然后重新利用這塊緩沖區(qū)。這個從內(nèi)存往磁盤寫數(shù)據(jù)的過程被稱為 Spill，中文可譯為溢寫，字面意思很直觀。這個溢寫是由單獨線程來完成，不影響往緩沖區(qū)寫 map 結(jié)果的線程。溢寫線程啟動時不應該阻止 map 的結(jié)果輸出，所以整個緩沖區(qū)有個溢寫的比例 spill.percent。這個比例默認是 0.8，也就是當緩沖區(qū)的數(shù)據(jù)已經(jīng)達到閾值（buffer size * spill percent = 100MB * 0.8 = 80MB），溢寫線程啟動，鎖定這 80MB 的內(nèi)存，執(zhí)行溢寫過程。Map task 的輸出結(jié)果還可以往剩下的 20MB 內(nèi)存中寫，互不影響。
當溢寫線程啟動后，需要對這 80MB 空間內(nèi)的 key 做排序 (Sort)。排序是 MapReduce 模型默認的行為，這里的排序也是對序列化的字節(jié)做的排序。
在這里我們可以想想，因為 map task 的輸出是需要發(fā)送到不同的 reduce 端去，而內(nèi)存緩沖區(qū)沒有對將發(fā)送到相同 reduce 端的數(shù)據(jù)做合并，那么這種合并應該是體現(xiàn)是磁盤文件中的。從官方圖上也可以看到寫到磁盤中的溢寫文件是對不同的 reduce 端的數(shù)值做過合并。所以溢寫過程一個很重要的細節(jié)在于，如果有很多個 key/value 對需要發(fā)送到某個 reduce 端去，那么需要將這些 key/value 值拼接到一塊，減少與 partition 相關的索引記錄。
在針對每個 reduce 端而合并數(shù)據(jù)時，有些數(shù)據(jù)可能像這樣：“aaa”/1， “aaa”/1。對于 WordCount 例子，就是簡單地統(tǒng)計單詞出現(xiàn)的次數(shù)，如果在同一個 map task 的結(jié)果中有很多個像“aaa”一樣出現(xiàn)多次的 key，我們就應該把它們的值合并到一塊，這個過程叫 reduce 也叫 combine。但 MapReduce 的術(shù)語中，reduce 只指 reduce 端執(zhí)行從多個 map task 取數(shù)據(jù)做計算的過程。除 reduce 外，非正式地合并數(shù)據(jù)只能算做 combine 了。其實大家知道的，MapReduce 中將 Combiner 等同于 Reducer。
如果 client 設置過 Combiner，那么現(xiàn)在就是使用 Combiner 的時候了。將有相同 key 的 key/value 對的 value 加起來，減少溢寫到磁盤的數(shù)據(jù)量。Combiner 會優(yōu)化 MapReduce 的中間結(jié)果，所以它在整個模型中會多次使用。那哪些場景才能使用 Combiner 呢？從這里分析，Combiner 的輸出是 Reducer 的輸入，Combiner 絕不能改變最終的計算結(jié)果。所以從我的想法來看，Combiner 只應該用于那種 Reduce 的輸入 key/value 與輸出 key/value 類型完全一致，且不影響最終結(jié)果的場景。比如累加，最大值等。Combiner 的使用一定得慎重，如果用好，它對 job 執(zhí)行效率有幫助，反之會影響 reduce 的最終結(jié)果。 每次溢寫會在磁盤上生成一個溢寫文件，如果 map 的輸出結(jié)果真的很大，有多次這樣的溢寫發(fā)生，磁盤上相應的就會有多個溢寫文件存在。當 map task 真正完成時，內(nèi)存緩沖區(qū)中的數(shù)據(jù)也全部溢寫到磁盤中形成一個溢寫文件。最終磁盤中會至少有一個這樣的溢寫文件存在 (如果 map 的輸出結(jié)果很少，當 map 執(zhí)行完成時，只會產(chǎn)生一個溢寫文件)，因為最終的文件只有一個，所以需要將這些溢寫文件歸并到一起，這個過程就叫做 Merge。Merge 是怎樣的？如前面的例子，“aaa”從某個 map task 讀取過來時值是 5，從另外一個 map 讀取時值是 8，因為它們有相同的 key，所以得 merge 成 group。什么是 group。對于“aaa”就是像這樣的：{“aaa”, [5, 8, 2, …]}，數(shù)組中的值就是從不同溢寫文件中讀取出來的，然后再把這些值加起來。請注意，因為 merge 是將多個溢寫文件合并到一個文件，所以可能也有相同的 key 存在，在這個過程中如果 client 設置過 Combiner，也會使用 Combiner 來合并相同的 key。
至此，map 端的所有工作都已結(jié)束，最終生成的這個文件也存放在 TaskTracker 夠得著的某個本地目錄內(nèi)。每個 reduce task 不斷地通過 RPC 從 JobTracker 那里獲取 map task 是否完成的信息，如果 reduce task 得到通知，獲知某臺 TaskTracker 上的 map task 執(zhí)行完成，Shuffle 的后半段過程開始啟動。
簡單地說，reduce task 在執(zhí)行之前的工作就是不斷地拉取當前 job 里每個 map task 的最終結(jié)果，然后對從不同地方拉取過來的數(shù)據(jù)不斷地做 merge，也最終形成一個文件作為 reduce task 的輸入文件。見下圖：
如 map 端的細節(jié)圖，Shuffle 在 reduce 端的過程也能用圖上標明的三點來概括。當前 reduce copy 數(shù)據(jù)的前提是它要從 JobTracker 獲得有哪些 map task 已執(zhí)行結(jié)束，這段過程不表，有興趣的朋友可以關注下。Reducer 真正運行之前，所有的時間都是在拉取數(shù)據(jù)，做 merge，且不斷重復地在做。如前面的方式一樣，下面我也分段地描述 reduce 端的 Shuffle 細節(jié)： Copy 過程，簡單地拉取數(shù)據(jù)。Reduce 進程啟動一些數(shù)據(jù) copy 線程 (Fetcher)，通過 HTTP 方式請求 map task 所在的 TaskTracker 獲取 map task 的輸出文件。因為 map task 早已結(jié)束，這些文件就歸 TaskTracker 管理在本地磁盤中。 Merge 階段。這里的 merge 如 map 端的 merge 動作，只是數(shù)組中存放的是不同 map 端 copy 來的數(shù)值。Copy 過來的數(shù)據(jù)會先放入內(nèi)存緩沖區(qū)中，這里的緩沖區(qū)大小要比 map 端的更為靈活，它基于 JVM 的 heap size 設置，因為 Shuffle 階段 Reducer 不運行，所以應該把絕大部分的內(nèi)存都給 Shuffle 用。這里需要強調(diào)的是，merge 有三種形式：1) 內(nèi)存到內(nèi)存 2) 內(nèi)存到磁盤 3) 磁盤到磁盤。默認情況下第一種形式不啟用，讓人比較困惑，是吧。當內(nèi)存中的數(shù)據(jù)量到達一定閾值，就啟動內(nèi)存到磁盤的 merge。與 map 端類似，這也是溢寫的過程，這個過程中如果你設置有 Combiner，也是會啟用的，然后在磁盤中生成了眾多的溢寫文件。第二種 merge 方式一直在運行，直到?jīng)]有 map 端的數(shù)據(jù)時才結(jié)束，然后啟動第三種磁盤到磁盤的 merge 方式生成最終的那個文件。 Reducer 的輸入文件。不斷地 merge 后，最后會生成一個“最終文件”。為什么加引號？因為這個文件可能存在于磁盤上，也可能存在于內(nèi)存中。對我們來說，當然希望它存放于內(nèi)存中，直接作為 Reducer 的輸入，但默認情況下，這個文件是存放于磁盤中的。至于怎樣才能讓這個文件出現(xiàn)在內(nèi)存中，之后的性能優(yōu)化篇我再說。當 Reducer 的輸入文件已定，整個 Shuffle 才最終結(jié)束。然后就是 Reducer 執(zhí)行，把結(jié)果放到 HDFS 上。
上面就是整個 Shuffle 的過程。細節(jié)很多，我很多都略過了，只試著把要點說明白。當然，我可能也有理解或表述上的很多問題，不吝指點。我希望不斷地完善和修改這篇文章，能讓它通俗、易懂，看完就能知道 Shuffle 的方方面面。至于具體的實現(xiàn)原理，各位有興趣就自己去探索，如果不方便的話，留言給我，我再來研究并反饋。到此，關于“MapReduce Shuffle 過程是怎樣的”的學習就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關知識，請繼續(xù)關注丸趣 TV 網(wǎng)站，丸趣 TV 小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
