原文作者：黎先生

原文地址：MapReduce基本原理及應(yīng)用

目錄

一、MapReduce模型簡(jiǎn)介

1. Map和Reduce函數(shù)

2. MapReduce體系結(jié)構(gòu)

3. MapReduce工作流程

4. MapReduce應(yīng)用程序執(zhí)行過(guò)程

二 、WordCount運(yùn)行實(shí)例

1.?WordCount的Map過(guò)程

2. WordCount的Reduce過(guò)程

3. WordCount源碼

---

一、MapReduce模型簡(jiǎn)介

　　MapReduce將復(fù)雜的、運(yùn)行于大規(guī)模集群上的并行計(jì)算過(guò)程高度地抽象到了兩個(gè)函數(shù)：Map和Reduce。它采用“分而治之”策略，一個(gè)存儲(chǔ)在分布式文件系統(tǒng)中的大規(guī)模數(shù)據(jù)集，會(huì)被切分成許多獨(dú)立的分片（split），這些分片可以被多個(gè)Map任務(wù)并行處理

1. Map和Reduce函數(shù)

Map和Reduce

2. MapReduce體系結(jié)構(gòu)

　　MapReduce體系結(jié)構(gòu)主要由四個(gè)部分組成，分別是：Client、JobTracker、TaskTracker以及Task

　　1）Client

　　用戶(hù)編寫(xiě)的MapReduce程序通過(guò)Client提交到JobTracker端 用戶(hù)可通過(guò)Client提供的一些接口查看作業(yè)運(yùn)行狀態(tài)

　　2）JobTracker

　　JobTracker負(fù)責(zé)資源監(jiān)控和作業(yè)調(diào)度 JobTracker 監(jiān)控所有TaskTracker與Job的健康狀況，一旦發(fā)現(xiàn)失敗，就將相應(yīng)的任務(wù)轉(zhuǎn)移到其他節(jié)點(diǎn) JobTracker 會(huì)跟蹤任務(wù)的執(zhí)行進(jìn)度、資源使用量等信息，并將這些信息告訴任務(wù)調(diào)度器（TaskScheduler），而調(diào)度器會(huì)在資源出現(xiàn)空閑時(shí)，

　　選擇合適的任務(wù)去使用這些資源

　　3）TaskTracker

　　TaskTracker 會(huì)周期性地通過(guò)“心跳”將本節(jié)點(diǎn)上資源的使用情況和任務(wù)的運(yùn)行進(jìn)度匯報(bào)給JobTracker，同時(shí)接收J(rèn)obTracker 發(fā)送過(guò)來(lái)的命令并執(zhí)行相應(yīng)的操作（如啟動(dòng)新任務(wù)、殺死任務(wù)等） TaskTracker 使用“slot”等量劃分本節(jié)點(diǎn)上的資源量（CPU、內(nèi)存等）。一個(gè)Task 獲取到

　　一個(gè)slot 后才有機(jī)會(huì)運(yùn)行，而Hadoop調(diào)度器的作用就是將各個(gè)TaskTracker上的空閑slot分配給Task使用。slot 分為Map slot 和Reduce slot 兩種，分別供MapTask 和Reduce Task 使用

　　4）Task

　　Task 分為Map Task 和Reduce Task 兩種，均由TaskTracker 啟動(dòng)

3. MapReduce工作流程

　　1) 工作流程概述

　

• 不同的Map任務(wù)之間不會(huì)進(jìn)行通信
• 不同的Reduce任務(wù)之間也不會(huì)發(fā)生任何信息交換
• 用戶(hù)不能顯式地從一臺(tái)機(jī)器向另一臺(tái)機(jī)器發(fā)送消息
• 所有的數(shù)據(jù)交換都是通過(guò)MapReduce框架自身去實(shí)現(xiàn)的

　　2) MapReduce各個(gè)執(zhí)行階段

　

4. MapReduce應(yīng)用程序執(zhí)行過(guò)程

　

二 、WordCount運(yùn)行實(shí)例

工作流程是Input從HDFS里面并行讀取文本中的內(nèi)容，經(jīng)過(guò)MapReduce模型，最終把分析出來(lái)的結(jié)果用Output封裝，持久化到HDFS中

1.?WordCount的Map過(guò)程

使用三個(gè)Map任務(wù)并行讀取三行文件中的內(nèi)容，對(duì)讀取的單詞進(jìn)行map操作，每個(gè)單詞都以<key, value>形式生成

　　

Map端源碼：

public class WordMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); StringTokenizer itr = new StringTokenizer(line); while (itr.hasMoreTokens()) { word.set(itr.nextToken().toLowerCase()); context.write(word, one); } } }

?

2. WordCount的Reduce過(guò)程

Reduce操作是對(duì)Map的結(jié)果進(jìn)行排序、合并等操作最后得出詞頻

　

Reduce端源碼

public class WordReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, new IntWritable(sum)); } }

3. WordCount源碼

import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class WordCount { public static class WordMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); StringTokenizer itr = new StringTokenizer(line); while (itr.hasMoreTokens()) { word.set(itr.nextToken().toLowerCase()); context.write(word, one); } } } public static class WordReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, new IntWritable(sum)); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); String[] otherArgs = new GenericOptionsParser(conf, args) .getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: wordcount <in> <out>"); System.exit(2); } Job job = new Job(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(WordMapper.class); job.setCombinerClass(WordReducer.class); job.setReducerClass(WordReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }

總結(jié)

以上是生活随笔為你收集整理的分布式离线计算—MapReduce—基本原理的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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