三、MapReduce的shuffle工作過程

[TOC]

一、基本流程

1、流程

map端：

1)假設(shè)有兩個map task并行運行。
2）每個map task任務(wù)處理完成后，會通過收集器 collector，將輸出結(jié)果存入一個環(huán)形緩沖區(qū)中。
3）環(huán)形緩沖區(qū)工作原理：

1>環(huán)形緩沖區(qū)默認(rèn)大小為100M，可以配置 
mapred-site.xml:mapreduce.task.io.sort.mb  來配置大小

2>環(huán)形緩沖區(qū)閾值為 80%，超過就會開始spill溢寫。可以配置
mapred-site.xml:mapreduce.map.sort.spill.percent  來配置閾值的百分比

3>環(huán)形緩沖區(qū)存儲兩種數(shù)據(jù)，
一種是元數(shù)據(jù)：KV的分區(qū)號，key的起始位置，value的起始位置，value長度。每個元數(shù)據(jù)長度固定為4個int長度
一種是原始數(shù)據(jù)：存儲key和value原本的數(shù)據(jù)

4>在元數(shù)據(jù)和原始數(shù)據(jù)的起始點，會有一個分界線，用于區(qū)分兩種數(shù)據(jù)的存儲區(qū)域，然后兩者往相反方向開始寫入數(shù)據(jù)。

5>當(dāng)環(huán)形緩沖區(qū)超過80%時，會將這80%的數(shù)據(jù)鎖定，然后溢寫到磁盤中變成小文件，并且這個過程中，這80%的空間不能寫入數(shù)據(jù)（由后臺一個新的線程來執(zhí)行溢寫）。同時剩下的20%可以繼續(xù)寫入數(shù)據(jù)。直到溢寫結(jié)束，解除80%的空間鎖定。

4）spill：當(dāng)緩沖器空間超過80%時，一個后臺線程會啟動，開始溢寫成小文件，寫入磁盤。在這個過程中，會對緩沖區(qū)中的元數(shù)據(jù)根據(jù)先根據(jù)分區(qū)號（每個分區(qū)一個溢寫文件），然后同一分區(qū)內(nèi)根據(jù)key進行排序（這里的排序算法使用快速排序）。接著根據(jù)排好序的元數(shù)據(jù)，溢寫相應(yīng)的原始數(shù)據(jù)。最后得到已經(jīng)分區(qū)且分區(qū)內(nèi)已根據(jù)key排好序的溢寫文件。
同時在溢寫最后一步，可以加入combine過程（可選的）。

這里說到是將元數(shù)據(jù)進行排序，然后根據(jù)排好序的元數(shù)據(jù)溢寫相應(yīng)的原始數(shù)據(jù)。為什么要這樣呢？
因為排序過程中涉及到數(shù)據(jù)的移動，而原始數(shù)據(jù)比起元數(shù)據(jù)一般都要大，所以涉及到移動成本（包括內(nèi)存空間，cpu等成本）比較大。所以這里直接根據(jù)原始數(shù)據(jù)中的key來排序元數(shù)據(jù)，最后形成一個有序的元數(shù)據(jù)區(qū)域。最后只要依次根據(jù)元數(shù)據(jù)，從原始數(shù)據(jù)區(qū)域中讀取對應(yīng)位置的KV，就可以得到有序的原始數(shù)據(jù)。

5）歸并排序：當(dāng)溢寫完成后，一般會發(fā)生多次溢寫，從而生成多個已分區(qū)，并且區(qū)內(nèi)有序的溢寫文件。接著就是將同一分區(qū)的多個溢寫文件進行歸并排序，合成一個大的溢寫文件，且里面是有序的。這個過程中也可以加入combine過程（可選的）。其實這個歸并的過程是分多次進行的，不是一次性完成的。
6）最后將歸并完后的溢寫文件經(jīng)過壓縮寫入到磁盤中。至此shuffle在map端的流程已經(jīng)完成

reduce端：一個分區(qū)對應(yīng)一個reduce task
7）reducer中有的MRAppMaster線程定期詢問map task輸出結(jié)果的文件位置，mapper會在map結(jié)束后向MRAppMaster匯報信息，從而reducer得到map的狀態(tài)，得到map的結(jié)果文件的目錄。接著reduce會自動向多個map拉取同一分區(qū)的結(jié)果文件到本地。拉取過程中，會先將數(shù)據(jù)暫存到緩沖區(qū)中，默認(rèn)是100M，也是環(huán)形緩沖區(qū)。當(dāng)數(shù)據(jù)量大于緩沖區(qū)大小時，就會將數(shù)據(jù)寫入到磁盤中。
8）歸并排序merge：拉取完成后，會將多個結(jié)果文件進行歸并排序，最終合成一個大的有序文件。這個merge 的過程中，會涉及到數(shù)據(jù)的輸入和輸出是在哪的，比如是輸入輸出都是內(nèi)存中的，輸入內(nèi)存，輸出硬盤的；輸入硬盤，輸出也是硬盤的。方式不同，明顯性能肯定不同。這是MapReduce優(yōu)化的一個點
9）接著會進行分組group過程。將同一個key的鍵值對合并成（key, array）的形式。如：(king,1), (king,2) 合并成 (king,[1,2])。這里面可以自定義分組方式。
10）而后面接著的group操作，一個分組就只會調(diào)用一次reduce方法，而且默認(rèn)只會使用分組中的第一個KV作為reduce的輸入，剩余的KV不會做處理，直接丟棄。這里可以自定義分組類。

11）merge，group過程完畢后，就會每個KV調(diào)用一次reduce方法，最終reduce輸出。

2、shuffle過程中的重點過程

partition：分區(qū)
spill：溢寫
merge：遷移合并
sort：排序。有3次排序。分別是溢寫中的快速排序，將多個溢寫文件合并的歸并排序。以及在reduce端中將多個map的結(jié)果文件進行歸并排序。
combine：map端的初次合并，業(yè)務(wù)邏輯就是reduce，只不過是局部的，這個過程是可選的。但是可以作為優(yōu)化的點，因為可以減少reduce從map拉取數(shù)據(jù)的數(shù)據(jù)量。

3、map端merge后的溢出文件的結(jié)構(gòu)

（1）存儲結(jié)構(gòu)

在map端溢出的文件其實有兩部分，一部分是索引文件，一部分是數(shù)據(jù)本身。

索引文件：主要是記錄各個分區(qū)在數(shù)據(jù)文件中的偏移量。
數(shù)據(jù)文件：記錄了KV的長度，KV的數(shù)據(jù)。

（2）特點

其實由此可以看出，map端存儲多個分區(qū)的溢寫結(jié)果時，并不是獨立存儲到獨立的文件中的，而是存儲到同一個文件中，然后通過索引文件來標(biāo)識各個分區(qū)的數(shù)據(jù)在總的數(shù)據(jù)文件中的偏移量來讀取不同分區(qū)的數(shù)據(jù)。這個存儲方式的好處是，如果分區(qū)很多話，單獨存儲每個分區(qū)的數(shù)據(jù)，就會生成多個文件，占用多個hdfs的索引資源。而采用上述的方式，則只需讀取兩個文件。