概述

雖然現在都在說大內存時代，不過內存的發展怎么也跟不上數據的步伐吧。所以，我們就要想辦法減小數據量。這里說的減小可不是真的減小數據量，而是讓數據分散開來。分開存儲、分開計算。這就是 MapReduce 分布式的核心。

---

版權說明

著作權歸作者所有。
商業轉載請聯系作者獲得授權，非商業轉載請注明出處。
本文作者：Q-WHai
發表日期： 2016年5月10日
本文鏈接：http://blog.csdn.net/lemon_tree12138/article/details/51367732
來源：CSDN
更多內容：分類 >> 大數據之 Hadoop

---

目錄

文章目錄

• 概述
• 版權說明
• 目錄
• • @[toc]
• MapReduce 簡介
• MapReduce 原理
• WordCount 程序
• • 需求分析
  • 邏輯實現
  • • Mapper
    • Reducer
    • Client
  • 本地運行
  • 分布式運行
  • • 打包
    • 上傳源數據
    • 分布式運行
    • 結果視窗
• Ref
• 征集

MapReduce 簡介

要了解 MapReduce，首先要了解 MapReduce 的載體是什么。在 Hadoop 中，用于執行 MapReduce 任務的機器有兩個角色：一個是 JobTracker，另一個是 TaskTracker。JobTracker 是用于管理和調度工作的，TaskTracker 是用于執行工作的。一個 Hadoop 集群中只有一臺 JobTracker（當然在 Hadoop 2.x 中，一個 Hadoop 集群中可能有多個 JobTracker）。

---

MapReduce 原理

MapReduce 模型的精髓在于它的算法思想——分治。對于分治的過程可以參見我之前的一篇博客《大數據算法：對5億數據進行排序》。還有就是可以去學習一下排序算法中的歸并排序，在這個排序算法中就是基于分治思想的。
回歸正題，在 MapReduce 模型中，可以把分治的這一概念表現得淋漓盡致。在處理大量數據的時候（比如說 1 TB，你別說沒有這么多的數據，大公司這點數據也不算啥的），如果只是單純地依賴我們的硬件，就顯得有些力不從心了。首先我們的內存沒有那么大，如放在磁盤上處理，那么過多的 IO 操作無疑是一個死穴。聰明的 Google 工程師總是給我們這些渣渣帶來驚喜，他們想把了把這些數據分散到許多機器上，在這些機器上完成一些初步的計算，再經過一系列的匯總，最后在我們的機器上（Master/Namenode）統計結果。
要知道我們不可能把我們的數據分散到隨意的 N 臺機器上。那么我們就必須讓這些機器之間建立一種可靠的關聯，這樣的關聯形成了一個計算機集群。這樣我們的數據就可以分發到集群中的各個計算機上了。在 Hadoop 里這一操作可以通過 -put 這一指令實現，關于這一點在下面的操作過程中也有體現。
當數據被上傳到 Hadoop 的 HDFS 文件系統上之后，就可以通過 MapReduce 模型中的 Mapper 先將數據讀進內存，過程像下面這樣：

經過 Mapper 的處理，數據會變成這樣

好了，到了這里，Map 的過程就已經結束了。接下來就是 Reduce 的過程了。

可以看到這里有一個 conbin 的過程，這個過程，也可以沒有的。而有的時候是一定不能有的，在后面我們可以會單獨來說說這里的 conbin，不過不是本文的內容，就不詳述了。
這樣整個 MapReduce 過程就已經 over 了，下面看看具體的實現及測試結果吧。

---

WordCount 程序

關于 WordCount 的 MapReduce 計算模型可參見本人的在線繪圖工具：https://www.processon.com/view/572bf161e4b0739b929916ea

需求分析

現在有大量的文件

每個文件又有大量的單詞

要求統計每個單詞的詞頻

邏輯實現

Mapper

public static class CoreMapper extends Mapper<Object, Text, Text, IntWritable> {private static final IntWritable one = new IntWritable(1);private static Text label = new Text();@Overrideprotected void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context)throws IOException, InterruptedException {StringTokenizer tokenizer = new StringTokenizer(value.toString());while(tokenizer.hasMoreTokens()) {label.set(tokenizer.nextToken());context.write(label, one);}}}

Reducer

public static class CoreReducer extends Reducer<Text, IntWritable, Text, IntWritable> {private IntWritable count = new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values,Reducer<Text, IntWritable, Text, IntWritable>.Context context)throws IOException, InterruptedException {if (null == values) {return;}int sum = 0;for (IntWritable intWritable : values) {sum += intWritable.get();}count.set(sum);context.write(key, count);}}

Client

public class ComputerClient extends Configuration implements Tool {public static void main(String[] args) {ComputerClient client = new ComputerClient();args = new String[] {AppConstant.INPUT,AppConstant.OUTPUT};try {ToolRunner.run(client, args);} catch (Exception e) {e.printStackTrace();}}@Overridepublic Configuration getConf() {return this;}@Overridepublic void setConf(Configuration arg0) {}@Overridepublic int run(String[] args) throws Exception {Job job = new Job(getConf(), "ComputerClient-job");job.setJarByClass(CoreComputer.class);job.setMapperClass(CoreComputer.CoreMapper.class);job.setCombinerClass(CoreComputer.CoreReducer.class);job.setReducerClass(CoreComputer.CoreReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));return job.waitForCompletion(true) ? 0 : 1;} }

本地運行

關于本地運行沒什么好說的，就是在 Eclipse 里配置好運行參數或是直接在代碼里指定輸入輸出路徑。然后 Run　As　一個　Hadoop　程序即可。

分布式運行

在分布式運行 MapReduce 的過程中，主要有以下幾個步驟：

打包

上傳源數據

分布式運行

打包

在打包的過程中，可以使用命令行打包，也可以使用 Eclipse 自帶的 Export。在 Eclipse 的打包導出過程中，與打包導出一個 Java 的 jar 過程是一樣的。這里就不多說了。假設我們打成的 jar 包為： job.jar

上傳源數據

上傳源數據是指將本地的數據上傳到 HDFS 文件系統上。
在上傳源數據之前我們需要在 HDFS 上新建你需要上傳的目標路徑，然后使用下面的這條指令即可完成數據的上傳。

$ hadoop fs -mkdir <hdfs_input_path> $ hadoop fs -put <local_path> <hdfs_input_path>

如果這里之前你不進行創建目錄，上傳過程會因為找不到目錄而出現異常情況。
數據上傳完成后，這些數據會分布在你整個集群的 DataNode 上，而不只是在你的本地機器上了。

分布式運行

等上面的所有事情已經就緒，那么就可以使用下面的 hadoop 指令運行我們的 hadoop 程序。

$ hadoop jar job.jar <hdfs_input_path> <hdfs_output_path>

結果視窗

打開瀏覽器
這里是程序中執行的過程中，進度的變化情況

下面是程序執行完成時的網頁截圖

---

Ref

• 《Hadoop 實戰》

---

征集

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

總結

以上是生活随笔為你收集整理的从 WordCount 到 MapReduce 计算模型的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。