MapReduce原理及實例分析

在VM上搭建好LINUX虛擬機(jī)，并安裝配置好HADOOP2.2.0，我這里是單節(jié)點(diǎn)的偽分布式

在eclipse中安裝hadoop插件

對我們這種MR的新手而言，最好在本地有一個HADOOP運(yùn)行環(huán)境，這樣有許多好處：

如果我們每次寫完MR程序，都打成JAR包上傳至線上服務(wù)器上運(yùn)行，那么每次MR運(yùn)行的時間非常長，也許等待了許久，運(yùn)行結(jié)果和我們預(yù)期不一致，又得改程序重新來一邊，這會有一點(diǎn)痛苦！

在我們本地的HADOOP上運(yùn)行MR程序非?？?，就那么幾秒，更加重要的是，我們可以再

本地準(zhǔn)備輸入文件去測試MR的邏輯，這對調(diào)試/開發(fā)程序非常方便！

假設(shè)，我們有這樣的輸入文件：

cate-a spu-1 1

cate-a spu-1 2

cate-a spu-2 3

cate-a spu-2 4

cate-a spu-3 5

cate-a spu-3 6

cate-a spu-1 7

cate-a spu-4 8

cate-a spu-4 9

cate-a spu-1 8

...

我們希望得到分cate,分spu的總和，并且取分cate分spu的TOP3

如上圖示，大致描述了MAP/REDUCE的運(yùn)行流程：

• 輸入文件+InputFormat  提供給MAP

• 需要清楚提供給MAP的KEY1/VALUE1是什么？MAP準(zhǔn)備輸出的KEY2/VALUE2是什么？

• MAP輸出后，會進(jìn)行分區(qū)操作，也就是決定KEY2/VALUE2發(fā)到哪些reduce上

  
  

• 分區(qū)由job.setPartitionerClass決定

  
  

• 在同一個分區(qū)內(nèi)，會對KEY2進(jìn)行排序，依據(jù)是job.setSortComparatorClass，

  如果沒有設(shè)置則根據(jù)KEY的compareTo方法

  
  

• 接下來進(jìn)入分組階段，會構(gòu)造KEY3和VALUE迭代器

  
  

• 分組的依據(jù)是job.setGroupingComparatorClass，只要比較器比較的相同就在同一組

  
  

• KEY3/VALUE迭代器交給reduce方法處理

  
  

步驟：

• 自定義KEY
  

KEY應(yīng)該是可序列化，可比較的，只需要注意實現(xiàn)WritableComparable即可。

重點(diǎn)關(guān)注compareTo方法。

@Override
public int compareTo(Cate2SpuKey that) {
System.out.println("開始對KEY進(jìn)行排序...");
if(cate2.equals(that.getCate2())){
return spu.compareTo(that.getSpu());
}
return cate2.compareTo(that.getCate2());
}

• 分區(qū)

分區(qū)，是KEY的第一次比較，extends Partitioner 并提供getPartition即可。

這里根據(jù)cate分區(qū)。

• 分組
  

需要注意的是，分組類必須提供構(gòu)造方法，并且重載 

public int compare(WritableComparable w1, WritableComparable w2) 。這里根據(jù)cate,spu分組。

通過上述的，就可以取得分cate分spu的SUM（counts）值了。

通過eclipse hadoop插件，可以方便我們上傳測試文件到HDFS，可以瀏覽，刪除HDFS文件，更加方便的是，就像運(yùn)行普通JAVA程序一樣的運(yùn)行/調(diào)試MR程序（不在需要打成JAR包）,讓我們可以追蹤MR的每一步，非常方便進(jìn)行邏輯性測試~

那么怎么取分cate分spu的TOP3呢？

我們只需要把上一個MR的輸出文件，作為另一個MR的輸入，并且以cate+counts 為KEY ，以spu為VALUE，根據(jù)cate分區(qū)，分組，排序的話:cate相同情況下，根據(jù)counts倒序；

最后在reduce階段取TOP3即可。

@Override
protected void reduce(Cate2CountsKey key, Iterable<Text> values,
Reducer<Cate2CountsKey, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
System.out.println("reduce...");
System.out.println("VALUES迭代前... key:" + key.toString());
System.out.println("VALUES迭代前... key:" + key.getCounts());
int top = 3;
for(Text t : values){
if(top > 0){
System.out.println("VALUES迭代中... key:" + key.toString());
System.out.println("VALUES迭代中... key:" + key.getCounts());
context.write(new Text(key.getCate2() + "\t" + t.toString()), 
new Text(key.getCounts() 
+ ""));
top--;
}
}
System.out.println("reduce over...");
}

那么到現(xiàn)在，分組取TOP就完成了。

在上面例子中的取TOP3的MR中，我們是以cate+counts為KEY，spu為VALUE。

cate作為分區(qū)，分組的依據(jù)，排序根據(jù)同一個cate下counts倒序。如下圖所示：

那么reduce方法中的KEY是什么？

spu1,spu4,spu3...是VALUES中的，那么這個迭代器的對應(yīng)KEY是什么呢？

是cate+42嗎？還是其他？

在VALUES迭代過程中，這個KEY會變化嗎？

我們可以看下ECLIPSE中的控制臺打印輸出的內(nèi)容：

從打印上來看，可以分析出如下結(jié)論：

分組后，交給reduce方法處理的KEY是同一組的所有KEY的第一個KEY，并且在VALUES迭代過程中，KEY并不會重新NEW，而是利用SETTER反射的方式重新設(shè)置屬性值，這樣在VALUES迭代過程中取得的KEY都是與之對應(yīng)的KEY了。