首先要推薦一下：
http://www.alidata.org/archives/1470
阿里的大牛在上面的文章中比較詳細(xì)的介紹了shuffle過程中mapper和reduce的每個過程，強烈推薦先讀一下。
?
不過，上文沒有寫明一些實現(xiàn)的細(xì)節(jié)，比如：spill的過程，mapper生成文件的 partition是怎么做的等等，相信有很多人跟我一樣在看了上面的文章后還是有很多疑問，我也是帶著疑問花了很久的看了cdh4.1.0版本 shuffle的邏輯，整理成本文，為以后回顧所用。
?
?首先用一張圖展示下map的流程：
在上圖中，我們假設(shè)此次mapreduce有多個mapper和2個reducer，p0 p1分別代表該數(shù)據(jù)應(yīng)該分配到哪個reducer端。我將mapper的過程大致分為5個過程。
?
1.prepare Input。
Mapreduce程序都需要指定輸入文件，輸入的格式有很多種，最常見的是保存在hdfs 上的文本文件。在用戶提交job到j(luò)obtrack(ResourceManager)前的job就會根據(jù)用戶的輸入文件計算出需要多少mapper，多 少reducer，mapper的輸入InputSplit有多大，block塊名稱等。mapper在prepare input階段只需要根據(jù)inputFormat類型創(chuàng)建對應(yīng)的RecordReader打開對應(yīng)的inputSplit分片即可。如果job配置了 combiner還需初始化combiner。代碼見MapTask類run方法
?
2.mapper process
這里的mapper指用戶使用或自己繼承的mapper類，這也是所有初學(xué)mapreduce的同學(xué)首先看到的類。
-
<span?style=
???*?Called?once?for?each?key/value?pair?in?the?input?split.?Most?applications?
-
???*?should?override?this,?but?the?default?is?the?identity?function.?
-
???*/
(
)??
-
protected
void
throws
</span>??
可以看到mapper默認(rèn)的map方法就是取出key，value并放到context對象中。context對象包裝了一個內(nèi)存中的buf，下面會介紹。
-
<span?style=
public
void
throws
while
??}</span>??
run方法就是mapper實際運行的過程：不停的從context的inputSplit對象中取出keyvalue對，通過map方法處理再保存到context包裝的內(nèi)存buf中。
?
3.buffer in memery
key value在寫入context中后實際是寫入MapOutputBuffer類中。在第一個階段的初始化過程中，MapOutputBuffer類會根據(jù)配置文件初始化內(nèi)存buffer，我們來看下都有哪些參數(shù)：
-
<span?style=
??
-
final
float
float
0.8
final
int
);??
-
if
float
1.0
float
0.0
throw
new
if
)?!=?sortmb)?{??
-
throw
new
,??
-
class
class
),?job);</span>??
partition：mapper的數(shù)據(jù)需要分配到reduce端的個數(shù)，由用戶的job指定，默認(rèn)為1.
spillper：內(nèi)存buf使用到此比例就會觸發(fā)spill，將內(nèi)存中的數(shù)據(jù)flush成一個文件。默認(rèn)為0.8
sortmb：內(nèi)存buf的大小，默認(rèn)100MB
indexCacheMemoryLimit：內(nèi)存index的大小。默認(rèn)為1024*1024
sorter：對mapper輸出的key的排序，默認(rèn)是快排
?
內(nèi)存buffer比較復(fù)雜，貼一張圖介紹一下這塊內(nèi)存buf的結(jié)構(gòu)：
當(dāng)一對keyvalue寫入時首先會從wrap buf的右側(cè)開始往左寫，同時，會把一條keyvalue的meta信息（partition，keystart，valuestart）寫入到最左邊的 index區(qū)域。當(dāng)wrap buf大小達(dá)到spill的觸發(fā)比例后會block寫入，挖出一部分?jǐn)?shù)據(jù)開始spill，直到spill完成后才能繼續(xù)寫，不過寫入位置不會置零，而是類 似循環(huán)buf那樣，在spill掉數(shù)據(jù)后可以重復(fù)利用內(nèi)存中的buf區(qū)域。
?
這里單獨講一下partition：
-
<span?style=
??
-
public
void
throws
????}</span>??
在keyvalue對寫入MapOutputBuffer時會調(diào)用 partitioner.getPartition方法計算partition即應(yīng)該分配到哪個reducer，這里的partition只是在內(nèi)存的 buf的index區(qū)寫入一條記錄而已，和下一個部分的partition不一樣哦。看下默認(rèn)的partitioner：HashPartition
-
<span?style=
??
-
public
int
int
return
??}</span>??
HashPartition只是把key hash后按reduceTask的個數(shù)取模，
因此一般來說，不同的key分配到哪個reducer是隨即的！所以，reducer內(nèi)的所有數(shù)據(jù)是有序的，但reducer之間的數(shù)據(jù)卻是亂序的！要想數(shù)據(jù)整體排序，要不只設(shè)一個reducer，要不使用TotalOrderPartitioner！
?
4.Partition Sort Store
在第四步中，partition是和sort一起做的，負(fù)責(zé)Spill的線程在拿到一段內(nèi)存buf后會調(diào)用QuickSort的sort方法進行內(nèi)存中的快排。
-
<span?style=
this
,?mstart,?mend,?reporter);</span>??
排序的算法是先按keyvalue記錄的partition排序后按key的compare方法：
-
<span?style=
public
int
final
int
final
int
final
int
final
int
final
int
final
int
??
-
if
return
??
-
return
????}</span>??
因此，mapper輸出的keyvalue首先是按partition聚合。而我們?nèi)绻付╧ey的compare方法會在這里生效并進行排序。最后，一次spill的輸出文件類似下圖。
在對內(nèi)存中的buf排序后開始寫文件。
-
<span?style=
for
int
;?i?<?partitions;?++i)?{??
-
null
try
long
new
if
null
??
-
new
while
final
int
else
int
while
??
-
??
-
if
new
????????????}</span>??
如果job沒有定義combiner則直接寫文件，
如果有combiner則在這里進行combine。
在生成spill文件后還會將此次spillRecord的記錄寫在一個index文件中。
-
<span?style=
??????????spillRec.writeToFile(indexFilename,?job);</span>??
-
<span?style=
????????????spillRec.putIndex(rec,?i);</span>??
?
5.merge
當(dāng)mapper執(zhí)行完畢后，就進入merge階段。首先看下相關(guān)的配置參數(shù)：
-
<span?style=
int
);</span>??
mergeFactor：同時merge的文件數(shù)。
?
merge階段的目的是將多個spill生成的中間文件合并為一個輸出文件，這里的合并不同 于combiner，無論有沒有配置combiner這里的merge都會執(zhí)行。merge階段的輸出是一個數(shù)據(jù)文件 MapFinalOutputFile和一個index文件。看下相關(guān)代碼：
-
<span?style=
new
null
??
-
long
new
if
null
else
??????????}</span>??
說下merge的算法。每個spill生成的文件中keyvalue都是有序的，但不同的文 件卻是亂序的，類似多個有序文件的多路歸并算法。Merger分別取出需要merge的spillfile的最小的keyvalue，放入一個內(nèi)存堆中， 每次從堆中取出一個最小的值，并把此值保存到merge的輸出文件中。這里和hbase中scan的算法非常相似，在分布式系統(tǒng)中多路歸并排序真是當(dāng)紅小 生啊！
這里merge時不同的partition的key是不會比較的，只有相同的partition的keyvalue才會進行排序和合并。最后的輸出文件類似下圖。
如果用戶定義了combiner，在merge的過程中也會進行combine，因為雖然第 四步中combine過但那只是部分輸入的combine，在merge時仍然需要combine。這里有人問了，既然這里有combiner，為啥在 spill輸出時還要combine納，我認(rèn)為是因為每次combine都會大大減少輸出文件的大小，spill時就combine能減少一定的IO操 作。
?
在merge完后會把不同partition的信息保存進一個index文件以便之后reducer來拉自己部分的數(shù)據(jù)。
-
<span?style=
??
-
??????????spillRec.putIndex(rec,?parts);</span>??
最后，我們再對mapper過程中的要點總結(jié)一下：
1.對map輸出<key,value>的分區(qū)（partition）是在寫入內(nèi)存buf前就做好的了，方法是對key的hash。我們可以通過繼承Partitioner類自己實現(xiàn)分區(qū)，將自己想要的數(shù)據(jù)分到同一個reducer中。
2.寫入內(nèi)存buf速度是非常快的，但spill過程會block寫入。因此，對內(nèi)存buf相關(guān)參數(shù)的調(diào)優(yōu)是mapreduce調(diào)優(yōu)的重點之一。
3.對數(shù)據(jù)的排序是基于MapOutKey排序的，因此，我們可以重載對應(yīng)的方法實現(xiàn)customize的排序順序
4.combine在spill和merge中都是進行。多次的combine會減少mapreduce中的IO操作，如果使用得當(dāng)會很好的提高性能。但需要注意的是要深刻理解combine的意義，比如平均值就不適合用combine。
