前言

我們不可能一直沉浸在 WordCount 的成功運(yùn)行當(dāng)中，就像之前學(xué)習(xí) Java 或是其他編程語(yǔ)言不會(huì)著迷于 HelloWord 一樣。
前面的 WordCount 程序只有一個(gè) Mapper 和一個(gè) Reducer 參與，也就是說(shuō)只有一個(gè) Job 參與。而一個(gè) Job 在通常情況下是無(wú)法滿足實(shí)際的開發(fā)需求，我們需要有更多的 Job 參與其中，并貢獻(xiàn)自己的力量。在 MapReduce 模塊中，有三個(gè)主要的模式：鏈?zhǔn)侥Ｐ?#xff0c;并發(fā)模型和組合模型。
本文首先以最簡(jiǎn)單的鏈?zhǔn)侥Ｐ妥鳛榍腥?#xff0c;讓你了解在 Hadoop 中，多個(gè) MapReduce 是如何協(xié)作完成任務(wù)的。

---

版權(quán)說(shuō)明

著作權(quán)歸作者所有。
商業(yè)轉(zhuǎn)載請(qǐng)聯(lián)系作者獲得授權(quán)，非商業(yè)轉(zhuǎn)載請(qǐng)注明出處。
本文作者：Q-WHai
發(fā)表日期： 2016年6月18日
本文鏈接：https://qwhai.blog.csdn.net/article/details/51706229
來(lái)源：CSDN
更多內(nèi)容：分類 >> 大數(shù)據(jù)之 Hadoop

---

鏈?zhǔn)?MapReduce 模型

所謂鏈?zhǔn)侥Ｊ?#xff0c;看名字就應(yīng)該很好理解。這就是簡(jiǎn)單地把幾個(gè) Job 串聯(lián)起來(lái)就可以了。模型圖如下：

---

示例列舉

假設(shè)有 4 個(gè)文件：

android java hadoop python

內(nèi)容分別為：
android

android fragment android java android activity map array screen activity map

java

java code java eclipse java java java code java map java spring mysql jvm java

hadoop

map reduce ssh mapreduce map reduce map reduce map reduce map reduce map reduce

python

python pycharm java dict java

承接之前 WordCount 的例子，這里我們要演示多 MapReduce 的操作，那么就不能只是 WordCount，所以，這里加入一個(gè)新的需求，將單詞數(shù)以 5 個(gè)一組，分成 N 組。將同一組的單詞寫入同一行。
這樣我們就不得不再添加一個(gè)新的 Job 來(lái)處理這個(gè)問題。

---

邏輯實(shí)現(xiàn)

在邏輯實(shí)現(xiàn)中，我們省略了第一個(gè) Job，也就是 WordCount 這一個(gè)模塊，如果你想要測(cè)試運(yùn)行，可以先寫好 WordCount 程序。
下面的兩塊內(nèi)容，分別是詞頻分組的 Mapper 跟 Reducer，代碼細(xì)節(jié)不是關(guān)鍵，且代碼的難度不大。這里就不細(xì)述了。
FrequenciesMapper.java

public static class FrequenciesMapper extends Mapper<Object, Text, IntWritable, Text> {final int groupSize = 5;@Overrideprotected void map(Object key, Text value, Mapper<Object, Text, IntWritable, Text>.Context context)throws IOException, InterruptedException {StringTokenizer tokenizer = new StringTokenizer(value.toString());String keyword = tokenizer.nextToken();int wordCount = Integer.parseInt(tokenizer.nextToken());int riseCount = (wordCount + groupSize) - (wordCount + groupSize) % groupSize;Text valueText = new Text("[" + keyword + ":" + wordCount + "]");context.write(new IntWritable(riseCount), valueText);} }

FrequenciesReducer.java

public static class FrequenciesReducer extends Reducer<IntWritable, Text, IntWritable, Text> {@Overrideprotected void reduce(IntWritable key, Iterable<Text> values,Reducer<IntWritable, Text, IntWritable, Text>.Context context)throws IOException, InterruptedException {if (null == values) {return;}boolean firstFlag = true;StringBuffer buffer = new StringBuffer();for (Text text : values) {buffer.append((firstFlag ? "" : ", ") + text.toString());firstFlag = false;}context.write(key, new Text(buffer.toString()));} }

下面就是把兩個(gè)不同的 Job 串聯(lián)在一起的核心工作了。
ChainMRClient.java

public class ChainMRClient {private static String inputPath = "./input";private static String outputPath = "./output";public static void main(String[] args) throws Exception {ChainMRClient client = new ChainMRClient();if (args.length == 2) {inputPath = args[0];outputPath = args[1];}client.execute();}private void execute() throws Exception {String tmpOutputPath = outputPath + "_tmp";runWordCountJob(inputPath, tmpOutputPath);runFrequenciesJob(tmpOutputPath, outputPath);}private int runWordCountJob(String inputPath, String outputPath) throws Exception {Configuration configuration = new Configuration();( ... 此處省略 N 行 ... )return job.waitForCompletion(true) ? 0 : 1;}private int runFrequenciesJob(String inputPath, String outputPath) throws Exception {Configuration configuration = new Configuration();( ... 此處省略 N 行 ... )return job.waitForCompletion(true) ? 0 : 1;} }

省略部分可以參考之前的 WordCount 代碼。
在上面的代碼中，可以看到 execute() 方法中先計(jì)算了中間的輸出目錄。這樣，第一個(gè) Job 就可以將結(jié)果輸出到這個(gè)目錄下，而不是我們向程序傳入的 outputPath 了；然后，再把這個(gè)中間輸出目錄作為第二個(gè) Job 的輸入目錄。
因?yàn)槭谴?lián)，所以只要把這兩個(gè) Job 放在代碼中合適位置串聯(lián)就可以了。

---

結(jié)果展示

中間結(jié)果

中間結(jié)果也就是前面的 WordCount 的結(jié)果，如下：

activity 2 android 3 array 1 code 2 dict 1 eclipse 1 fragment 1 java 11 jvm 1 map 9 mapreduce 1 mysql 1 pycharm 1 python 1 reduce 6 screen 1 spring 1 ssh 1

最終結(jié)果

可以看到在最終的結(jié)果里，單詞已經(jīng)被統(tǒng)計(jì)好詞頻，并進(jìn)行了分組。如下：

5 [ssh:1], [spring:1], [screen:1], [python:1], [pycharm:1], [mysql:1], [mapreduce:1], [jvm:1], [fragment:1], [eclipse:1], [dict:1], [code:2], [array:1], [android:3], [activity:2] 10 [reduce:6], [map:9] 15 [java:11]

串聯(lián) Job

這里展示的就是剛剛運(yùn)行成功的兩個(gè) Job。

---

征集

如果你也需要使用ProcessOn這款在線繪圖工具，可以使用如下邀請(qǐng)鏈接進(jìn)行注冊(cè)：
https://www.processon.com/i/56205c2ee4b0f6ed10838a6d

總結(jié)

以上是生活随笔為你收集整理的MapReduce进阶：多MapReduce的链式模式的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。