Mapreduce shuffle的示例分析

這篇文章給大家分享的是有關(guān)Mapreduce shuffle的示例分析的內(nèi)容。小編覺(jué)得挺實(shí)用的，因此分享給大家做個(gè)參考，一起跟隨小編過(guò)來(lái)看看吧。

Mapreduce shuffle詳解

Mapreduce確保每個(gè)reducer的的輸入都是按鍵排序的。系統(tǒng)執(zhí)行排序的過(guò)程(即將map輸出作為輸入 傳給reducer)成為shuffle。從多個(gè)方面來(lái)看shuffle是mapreduce的心臟，是奇跡發(fā)生的地方。

上圖展示了，mapreduce的詳細(xì)過(guò)程。

1  輸入分片

對(duì)于數(shù)據(jù)的輸入分片，要根據(jù)不同的存儲(chǔ)格式有不同的介紹。對(duì)于，hdfs存儲(chǔ)的文件，數(shù)據(jù)的分片就可分為兩種，文件可切分(不壓縮或者壓縮格式bzip2)的按照一定大小進(jìn)行分片有既定算法，默認(rèn)是block的大小，具體算法不在這里細(xì)講，前面hive調(diào)優(yōu)的文章又說(shuō)到，而且浪尖也會(huì)在后續(xù)的文章提到這個(gè)內(nèi)容；

分片的時(shí)候計(jì)算公式計(jì)算過(guò)程舉例

文件不可切分則一個(gè)文件一個(gè)分片。

2  Map端

從上圖我們可以看到map端的處理過(guò)程。Map會(huì)讀取輸入分片數(shù)據(jù)。但是map函數(shù)開(kāi)始產(chǎn)生輸出時(shí)并不是簡(jiǎn)單的將數(shù)據(jù)寫(xiě)入磁盤。這個(gè)過(guò)程很復(fù)雜，他利用了緩沖的方式寫(xiě)到內(nèi)存并出于效率的考慮進(jìn)行排序。

每個(gè)map任務(wù)都是有一個(gè)環(huán)形緩沖區(qū)的用于存儲(chǔ)任務(wù)的輸出。在默認(rèn)情況下，緩沖區(qū)的大小為100MB，辭職可以通過(guò)改變io.sort.mb來(lái)調(diào)整。一旦緩沖內(nèi)容達(dá)到閾值(io.sort,spill,percent，默認(rèn)是0.8)，一個(gè)后臺(tái)線程會(huì)將內(nèi)容spill到磁盤。在spill到磁盤的過(guò)程中，map輸出并不會(huì)停止往緩沖區(qū)寫(xiě)入數(shù)據(jù)，但如果在此期間緩沖區(qū)被寫(xiě)滿，map會(huì)被阻塞知道寫(xiě)磁盤過(guò)程完成。

溢出寫(xiě)過(guò)程安裝輪詢方式將緩沖區(qū)的內(nèi)容寫(xiě)到mapred.local.dir指定的作業(yè)特定子目錄中的目錄中。

寫(xiě)磁盤之前，線程首先根據(jù)數(shù)據(jù)最終要傳的reducer把數(shù)據(jù)劃分成相應(yīng)的分區(qū)。在每個(gè)分區(qū)中后臺(tái)線程按鍵進(jìn)行內(nèi)排序，如果有一個(gè)combiner，它就在排序后的輸出上運(yùn)行。運(yùn)行combinner使得map輸出結(jié)果更緊湊，因此可以減少寫(xiě)到磁盤的數(shù)據(jù)和傳遞給reducer的數(shù)據(jù)。

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

如果至少存在3個(gè)溢出文件(通過(guò)min.num.spills.for.combine屬性設(shè)置)時(shí)，則combiner就會(huì)在輸出文件寫(xiě)到磁盤之前再次運(yùn)行。前面曾說(shuō)過(guò)，combiner可以在輸入上反復(fù)運(yùn)行，單不影響最終的結(jié)果。如果只有一兩個(gè)溢出文件，那么對(duì)map輸出的減少不值得調(diào)用combiner，就不會(huì)為map輸出再次運(yùn)行combiner。

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

Reducer是通過(guò)HTTP的方式得到輸出文件的分區(qū)。在MRV2中使用netty進(jìn)行數(shù)據(jù)傳輸，默認(rèn)情況下netty的工作線程數(shù)是處理器數(shù)的2倍。MRV1中，默認(rèn)值是40，由tracker.http.threads來(lái)在tasktracker端設(shè)定。

3   Reducer端

集群中往往一個(gè)mr任務(wù)會(huì)有若干map任務(wù)和reduce任務(wù)，map任務(wù)運(yùn)行有快有慢，reduce不可能等到所有的map任務(wù)都運(yùn)行結(jié)束再啟動(dòng)，因此只要有一個(gè)任務(wù)完成，reduce任務(wù)就開(kāi)始復(fù)制器輸出。復(fù)制線程的數(shù)量由mapred.reduce.parallel.copies屬性來(lái)改變，默認(rèn)是 5。

Reducer如何知道m(xù)ap輸出的呢？對(duì)于MRv2 map運(yùn)行結(jié)束之后直接就通知了appmaster，對(duì)于給定的job appmaster是知道m(xù)ap的輸出和host之間的關(guān)系。在reduce端獲取所有的map輸出之前，Reduce端的線程會(huì)周期性的詢問(wèn)master 關(guān)于map的輸出。Reduce并不會(huì)在獲取到map輸出之后就立即刪除hosts，因?yàn)閞educe有肯能運(yùn)行失敗。相反，是等待appmaster的刪除消息來(lái)決定刪除host。

Reduce對(duì)map輸出的不同大小也有相應(yīng)的調(diào)優(yōu)處理。如果map輸出相當(dāng)小，會(huì)被復(fù)制到reduce任務(wù)JVM的內(nèi)存(緩沖區(qū)大小由mapred.job.shuffle.input.buffer.percent屬性控制，指定用于此用途的堆空間的百分比)，否則，map輸出會(huì)被復(fù)制到磁盤。一旦內(nèi)存緩沖區(qū)達(dá)到閾值(由mapred.job.shuffle.merge.percent決定)或達(dá)到map的輸出閾值(mapred.inmem.merge,threshold控制)，則合并后溢出寫(xiě)到磁盤中。如果指定combiner，則在合并期間運(yùn)行它已降低寫(xiě)入磁盤的數(shù)據(jù)量。

隨著磁盤上副本的增多，后臺(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(默認(rèn)值是10，由io.sort.factor屬性設(shè)置，與map的合并類似)，合并將進(jìn)行5趟。每趟將10個(gè)文件合并成一個(gè)文件，因此最后有5個(gè)中間文件。

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

在reduce階段，對(duì)已排序輸出中的每個(gè)鍵調(diào)用reduce函數(shù)。此階段的輸出直接寫(xiě)到輸出文件系統(tǒng)，一般為hdfs。

注意：

每趟合并的文件數(shù)實(shí)際上比上面例子中展示的有所不同的。目標(biāo)是合并最小數(shù)據(jù)量的文件以便滿足最后一趟的合并系數(shù)。因此，如果有40個(gè)文件，我們不會(huì)再四趟中每趟合并10個(gè)文件而得到4個(gè)文件。相反，第一趟只合并4個(gè)文件，隨后的三塘合并10個(gè)文件。最后一趟中，4個(gè)已經(jīng)合并的文件和剩余的6個(gè)文件合計(jì)是個(gè)文件進(jìn)行合并。如下圖所述：

注意這并沒(méi)有改變合并的次數(shù)，它只是一個(gè)優(yōu)化措施，目的是盡量減少寫(xiě)到磁盤的數(shù)據(jù)量，因?yàn)樽詈笠惶丝偸侵苯雍喜⒌絩educe。

感謝各位的閱讀！關(guān)于“Mapreduce shuffle的示例分析”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，讓大家可以學(xué)到更多知識(shí)，如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到吧！