MapReduce是一種編程模型，用于大規(guī)模數(shù)據(jù)集(大于1TB)的并行運(yùn)算。MapReduce采用”分而治之”的思想，把對(duì)大規(guī)模數(shù)據(jù)集的操作，分發(fā)給一個(gè)主節(jié)點(diǎn)管理下的各個(gè)分節(jié)點(diǎn)共同完成，然后通過(guò)整合各個(gè)節(jié)點(diǎn)的中間結(jié)果，得到最終結(jié)果。簡(jiǎn)單地說(shuō)，MapReduce就是”任務(wù)的分解與結(jié)果的匯總”。

MapReduce架構(gòu)

先來(lái)看一下MapReduce1.0的架構(gòu)圖

上圖中的TaskTracker對(duì)應(yīng)HDFS中的DataNode，

在MapReduce1.x中，用于執(zhí)行MapReduce任務(wù)的機(jī)器角色有兩個(gè)：一個(gè)是JobTracker;另一個(gè)是TaskTracker，JobTracker是用于調(diào)度工作的，TaskTracker是用于執(zhí)行工作的。一個(gè)Hadoop集群中只有一臺(tái)JobTracker。

流程分析

1. 在客戶端啟動(dòng)任務(wù)，客戶端向JobTracker請(qǐng)求一個(gè)Job ID。
2. 將運(yùn)行任務(wù)所需要的程序文件復(fù)制到HDFS上，包括MapReduce程序打包的JAR文件、配置文件和客戶端計(jì)算所得的輸入劃分信息。這些文件都存放在JobTracker專門為該任務(wù)創(chuàng)建的文件夾中。文件夾名Job ID。
3. JobTracker接收到任務(wù)后，將其放在一個(gè)隊(duì)列里，等待調(diào)度器對(duì)其進(jìn)行調(diào)度，當(dāng)作業(yè)調(diào)度器根據(jù)自己的調(diào)度算法調(diào)度到該任務(wù)時(shí)，會(huì)根據(jù)輸入劃分信息創(chuàng)建N個(gè)map任務(wù)，并將map任務(wù)分配給N個(gè)TaskTracker(DataNode)執(zhí)行。
4. map任務(wù)不是隨隨便便地分配給某個(gè)TaskTracker的，這里有個(gè)概念叫：數(shù)據(jù)本地化(Data-Local)。意思是：將map任務(wù)分配給含有該map處理的數(shù)據(jù)塊的TaskTracker上，同時(shí)將程序JAR包復(fù)制到該TaskTracker上來(lái)運(yùn)行，這叫“運(yùn)算移動(dòng)，數(shù)據(jù)不移動(dòng)”。而分配reduce任務(wù)時(shí)并不考慮數(shù)據(jù)本地化。
5. TaskTracker每隔一段時(shí)間會(huì)給JobTracker發(fā)送一個(gè)Heartbeat(心跳)，告訴JobTracker它依然在運(yùn)行，同時(shí)心跳中還攜帶著很多的信息，比如當(dāng)前map任務(wù)完成的進(jìn)度等信息。當(dāng)JobTracker收到作業(yè)的***一個(gè)任務(wù)完成信息時(shí)，便把該作業(yè)設(shè)置成“成功”。當(dāng)JobClient查詢狀態(tài)時(shí)，它將得知任務(wù)已完成，便顯示一條消息給用戶。

以上是在客戶端、JobTracker、TaskTracker的層次來(lái)分析MapReduce的工作原理的，下面我們?cè)偌?xì)致一點(diǎn)，從map任務(wù)和reduce任務(wù)的層次來(lái)分析分析吧。

MapReduce運(yùn)行流程

以wordcount為例，運(yùn)行的詳細(xì)流程圖如下

1.split階段

首先mapreduce會(huì)根據(jù)要運(yùn)行的大文件來(lái)進(jìn)行split，每個(gè)輸入分片(input split)針對(duì)一個(gè)map任務(wù)，輸入分片(input split)存儲(chǔ)的并非數(shù)據(jù)本身，而是一個(gè)分片長(zhǎng)度和一個(gè)記錄數(shù)據(jù)位置的數(shù)組。輸入分片(input split)往往和HDFS的block(塊)關(guān)系很密切，假如我們?cè)O(shè)定HDFS的塊的大小是64MB，我們運(yùn)行的大文件是64x10M，mapreduce會(huì)分為10個(gè)map任務(wù)，每個(gè)map任務(wù)都存在于它所要計(jì)算的block(塊)的DataNode上。

2.map階段

map階段就是程序員編寫的map函數(shù)了，因此map函數(shù)效率相對(duì)好控制，而且一般map操作都是本地化操作也就是在數(shù)據(jù)存儲(chǔ)節(jié)點(diǎn)上進(jìn)行。本例的map函數(shù)如下：

publicclassWCMapperextendsMapperLongWritable,Text,Text,IntWritable{@Override  
protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{  
Stringstr=value.toString();  
String[]strs=StringUtils.split(str,'');for(Strings:strs){  
context.write(newText(s),newIntWritable(1));  
}  
}  
} 

根據(jù)空格切分單詞，計(jì)數(shù)為1，生成key為單詞，value為出現(xiàn)1次的map供后續(xù)計(jì)算。

3.shuffle階段

shuffle階段主要負(fù)責(zé)將map端生成的數(shù)據(jù)傳遞給reduce端，因此shuffle分為在map端的過(guò)程和在reduce端的執(zhí)行過(guò)程。

先看map端：

1. map首先進(jìn)行數(shù)據(jù)結(jié)果數(shù)據(jù)屬于哪個(gè)partition的判斷，其中一個(gè)partition對(duì)應(yīng)一個(gè)reduce，一般通過(guò)key.hash()%reduce個(gè)數(shù)來(lái)實(shí)現(xiàn)。
2. 把map數(shù)據(jù)寫入到Memory Buffer(內(nèi)存緩沖區(qū))，到達(dá)80%閥值，開啟溢寫進(jìn)磁盤過(guò)程，同時(shí)進(jìn)行key排序，如果有combiner步驟，則會(huì)對(duì)相同的key做歸并處理，最終多個(gè)溢寫文件合并為一個(gè)文件。

reduce端：

reduce節(jié)點(diǎn)從各個(gè)map節(jié)點(diǎn)拉取存在磁盤上的數(shù)據(jù)放到Memory Buffer(內(nèi)存緩沖區(qū))，同理將各個(gè)map的數(shù)據(jù)進(jìn)行合并并存到磁盤，最終磁盤的數(shù)據(jù)和緩沖區(qū)剩下的20%合并傳給reduce階段。

4.reduce階段

reduce對(duì)shuffle階段傳來(lái)的數(shù)據(jù)進(jìn)行***的整理合并

publicclassWCReducerextendsReducerText,IntWritable,Text,IntWritable{@Override  
protectedvoidreduce(Textkey,IterableIntWritablevalues,Contextcontext)throwsIOException,InterruptedException{intsum=0;for(IntWritablei:values){  
sum+=i.get();  
}  
context.write(key,newIntWritable(sum));  
}  
} 

MapReduce的優(yōu)缺點(diǎn)

優(yōu)點(diǎn)：

1. 易于編程;
2. 良好的擴(kuò)展性;
3. 高容錯(cuò)性;

4.適合PB級(jí)別以上的大數(shù)據(jù)的分布式離線批處理。

缺點(diǎn)：

1. 難以實(shí)時(shí)計(jì)算(MapReduce處理的是存儲(chǔ)在本地磁盤上的離線數(shù)據(jù))
2. 不能流式計(jì)算(MapReduce設(shè)計(jì)處理的數(shù)據(jù)源是靜態(tài)的)
3. 難以DAG計(jì)算MapReduce這些并行計(jì)算大都是基于非循環(huán)的數(shù)據(jù)流模型，也就是說(shuō)，一次計(jì)算過(guò)程中，不同計(jì)算節(jié)點(diǎn)之間保持高度并行，這樣的數(shù)據(jù)流模型使得那些需要反復(fù)使用一個(gè)特定數(shù)據(jù)集的迭代算法無(wú)法高效地運(yùn)行。