MapReduce的Shuffle機(jī)制是什么

Shuffle過(guò)程，也稱Copy階段。reduce task從各個(gè)map task上遠(yuǎn)程拷貝一片數(shù)據(jù)，并針對(duì)某一片數(shù)據(jù)，如果其大小超過(guò)一定的閥值，則寫(xiě)到磁盤(pán)上，否則直接放到內(nèi)存中。

MAP端

map函數(shù)開(kāi)始產(chǎn)生輸出時(shí)，并不是簡(jiǎn)單地將它寫(xiě)到磁盤(pán)上。這個(gè)過(guò)程更復(fù)雜，它利用緩沖的方式寫(xiě)到內(nèi)存并出于效率的目的進(jìn)行預(yù)排序。

每個(gè)map任務(wù)都有一個(gè)環(huán)形緩沖區(qū)用于存儲(chǔ)任務(wù)輸出。在默認(rèn)情況下，緩沖區(qū)的大小為100MB，這個(gè)值可以通過(guò)mapreduce.task.io.sort.mb屬性來(lái)調(diào)整。一旦緩沖內(nèi)容達(dá)到閾值（mapreduce.map.sort.spill.percent，默認(rèn)為80%），一個(gè)后臺(tái)線程便開(kāi)始把內(nèi)容溢寫(xiě)（spill）到磁盤(pán)，在溢寫(xiě)到磁盤(pán)的過(guò)程中，map輸出繼續(xù)寫(xiě)道緩沖區(qū)，但如果在此期間緩沖區(qū)被寫(xiě)滿，map會(huì)被阻塞直到磁盤(pán)過(guò)程完成。溢寫(xiě)過(guò)程按輪詢方式將緩沖區(qū)的內(nèi)容寫(xiě)到mapreduce.cluster.local.dir屬性在作業(yè)特定子目錄下的指定的目錄中。在寫(xiě)磁盤(pán)之前，線程首先根據(jù)數(shù)據(jù)最終要傳的reducer把數(shù)據(jù)劃分成相應(yīng)的分區(qū)（partition，用戶也可自定義分區(qū)函數(shù)，但默認(rèn)的partitioner通過(guò)哈希函數(shù)來(lái)分區(qū)，也很高效）。在每個(gè)分區(qū)中，后臺(tái)線程按鍵進(jìn)行內(nèi)存中排序，如果有一個(gè)combiner函數(shù)，它就在排序后的輸出上運(yùn)行。運(yùn)行combiner函數(shù)使得map輸出結(jié)果更緊湊，因此減少寫(xiě)到磁盤(pán)的數(shù)據(jù)和傳遞給reducer的數(shù)據(jù)。

每次內(nèi)存緩沖區(qū)達(dá)到溢出閾值時(shí)，就會(huì)新建一個(gè)溢出文件（spill file），因此，在map任務(wù)寫(xiě)完其最后一個(gè)輸出記錄后，會(huì)有幾個(gè)溢寫(xiě)文件。在任務(wù)完成之前，溢寫(xiě)文件被合并成一個(gè)已分區(qū)且已排序的輸出文件。配置屬性是mapreduce.task.io.sort.factor控制著一次最多能合并多少流，默認(rèn)值是10.

如果至少存在3個(gè)溢寫(xiě)文件（通過(guò)mapreduce.map.combine.minspills屬性設(shè)置）時(shí)，則combiner就會(huì)在輸出文件寫(xiě)到磁盤(pán)之前再次運(yùn)行。combiner可以在輸入上反復(fù)運(yùn)行，但并不影響最終結(jié)果。如果只有1個(gè)或者2個(gè)溢寫(xiě)文件，那么由于map輸出規(guī)模減少，因此不值得調(diào)用combiner帶來(lái)的開(kāi)銷(xiāo)，因此不會(huì)為該map輸出再次運(yùn)行combiner。

在將壓縮map輸出寫(xiě)到磁盤(pán)的過(guò)程中對(duì)他進(jìn)行壓縮往往是一個(gè)很好的主意，因?yàn)檫@樣寫(xiě)磁盤(pán)的速度更快，節(jié)約磁盤(pán)空間，并且減少傳給reducer的數(shù)據(jù)量。在默認(rèn)情況下，輸出時(shí)不壓縮的，但只要將mapreduce.map.output.compress設(shè)置為true，就可以輕松使用此功能。使用的壓縮庫(kù)由mapreduce.map.output.compress.codec指定。

reducer通過(guò)HTTP得到輸出文件的分區(qū)。用于文件分區(qū)的工作線程的數(shù)量由任務(wù)的mapreduce.shuffle.max.threads屬性控制，此設(shè)置針對(duì)的是每一個(gè)節(jié)點(diǎn)管理器，而不是針對(duì)每個(gè)map任務(wù)。默認(rèn)值0將最大線程數(shù)設(shè)置為機(jī)器中處理器數(shù)量的兩倍。

REDUCE端

現(xiàn)在轉(zhuǎn)到處理過(guò)程的reduce部分。map輸出文件位于運(yùn)行map任務(wù)的tasktracker的本地磁盤(pán)（注意，盡管map輸出經(jīng)常寫(xiě)到map tasktracker 的本地磁盤(pán)，但reduce輸出并不這樣），現(xiàn)在，tasktracker需要為分區(qū)文件運(yùn)行reduce任務(wù)。并且，reduce任務(wù)需要集群上若干個(gè)map任務(wù)的map輸出作為其特殊的分區(qū)文件。每個(gè)map任務(wù)的完成時(shí)間可能不同，因此在每個(gè)任務(wù)完成時(shí)，reduce任務(wù)就開(kāi)始復(fù)制其輸出。這就是reduce任務(wù)的復(fù)制階段。reduce任務(wù)有少量復(fù)制線程，因此能夠并行取得map輸出。默認(rèn)值是5個(gè)線程，但這個(gè)默認(rèn)值可以修改設(shè)置mapreduce.reduce.shuffle.parallelcopies屬性即可。

如果map輸出相當(dāng)小，會(huì)被復(fù)制到reduce任務(wù)JVM的內(nèi)存（緩沖區(qū)大小由mapreduce.reduce.shuffle.input.buffer.percent屬性控制，指定用于此用途的堆空間的百分比），否則，map輸出被復(fù)制到磁盤(pán)。一旦內(nèi)存緩沖區(qū)達(dá)到閾值大小（由mapreduce.reduce.shuffle.merge.percent決定）或者達(dá)到map輸出閾值（由mapreduce.reduce.merge.inmen.threshold控制），則合并后溢出寫(xiě)到磁盤(pán)中。如果指定combiner，則在合并期間運(yùn)行它以降低寫(xiě)入硬盤(pán)的數(shù)據(jù)量。

隨著磁盤(pán)上副本增多，后臺(tái)線程會(huì)將它們合并為更大的、排好序的文件。這會(huì)為后面的合并節(jié)省一些時(shí)間。注意，為了合并，壓縮的map輸出（通過(guò)map任務(wù)）都必須在內(nèi)存中被解壓縮。

復(fù)制完所有map輸出后，reduce任務(wù)進(jìn)入排序階段（更恰當(dāng)?shù)恼f(shuō)法是合并階段，因?yàn)榕判蚴窃趍ap端進(jìn)行的），這個(gè)階段將合并map輸出，維持其順序排序。這是循環(huán)進(jìn)行的。比如，如果有50個(gè)map輸出，而合并因子是10（10為默認(rèn)設(shè)置，由mapreduce.task.io.sort.factor屬性設(shè)置，與map的合并類(lèi)似），合并將進(jìn)行5趟，每趟將10個(gè)文件合并成一個(gè)文件，因此最后有5個(gè)中間文件。

在最后階段，即reduce階段，直接把數(shù)據(jù)輸入reduce函數(shù)，從而省略了一次磁盤(pán)往返行程，并沒(méi)有將這5個(gè)文件合并成一個(gè)已排序的文件作為最后一趟。最后的合并可以來(lái)自內(nèi)存和磁盤(pán)片段。

?

每趟合并的文件數(shù)實(shí)際上比事例中展示有所不同。目標(biāo)是合并最少數(shù)量的文件以便滿足于最后一趟的合并系數(shù)。因此如果有40個(gè)文件，我們并不會(huì)在四趟中每趟合并10個(gè)文件從而得到4個(gè)文件。相反，第一趟只合并4個(gè)文件，隨后的三趟合并完整的10個(gè)文件。在最后一趟中，4個(gè)已合并的文件和余下的6個(gè)（未合并的）文件合計(jì)10個(gè)。

在reduce階段，對(duì)已排序輸出中的每個(gè)鍵都調(diào)用reduce函數(shù)。此階段的輸出直接寫(xiě)到輸出文件系統(tǒng)，一般為HDFS（可自定義）。如果采用HDFS，由于節(jié)點(diǎn)管理器也運(yùn)行數(shù)據(jù)節(jié)點(diǎn)，所以第一個(gè)塊的副本將被寫(xiě)入到本地磁盤(pán)。