Hadoop越来越火&＃xff0c;而Hadoop里面有个核心的玩意&＃xff0c;那就是MapReduce&＃xff0c;它在Hadoop的并行计算中承担很重要的作用&＃xff0c;也是在Hadoop下做程序开发时&＃xff0c;必须要了解的&＃xff0c;下面我们就MapRecude的一个简单例子WordCount来做一下深入的了解和分析。
先跟远哥一起先了解一下什么是MapReduce吧。
首先MapReduce它是两个英文单词组成的&＃xff0c;Map表示映射&＃xff0c;Reduce表示化简&＃xff0c;它是一种编程模型&＃xff0c;用于大规模数据集(大于1TB)的并行运算&＃xff0c;主要思想来自函数式编程。
在Hadoop中&＃xff0c;MapReduce过程分三个步骤&＃xff1a;Map(主要是分解并行的任务)、Combine(主要是为了提高Reduce的效率)和Reduce(把处理后的结果再汇总起来) 。
好了&＃xff0c;我们先看一下运行一个Hadoop作业的启动代码&＃xff1a;
Job job &＃61;
new Job(conf, "word count");
job.setJarByClass(WordCount.
class);
job.setMapperClass(
TokenizerMapper.
class);
job.setCombinerClass(
IntSumReducer.
class);
job.setReducerClass(
IntSumReducer.
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);
看到没&＃xff0c;在运行一个Hadoop作业之前&＃xff0c;先得指定 MapperClass 、CombinerClass 、ReducerClass .
假设我们交给Hadoop去分析的一个文本内容为&＃xff1a;
lixy csy lixy zmde nitamade hehe
realy amoeba woyou weibo hehe
好了&＃xff0c;提供的内容很简单&＃xff0c;就是3行文本&＃xff0c;第1行文本包含n个单词&＃xff0c;第2行是空的&＃xff0c;第3行也包含n个单词&＃xff0c;单词与单词之间用空格隔开&＃xff0c;下面我们来看看MapperClass 是如何实现的&＃xff0c;又是如何运行的呢&＃xff1f;看看 TokenizerMapper 的代码&＃xff1a;
public
class TokenizerMapper
extends Mapper {
private
final
static IntWritable one &＃61;
new IntWritable(1);
private Text word &＃61;
new Text();
public
void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
System.out.println(
"TokenizerMapper.map...");
System.out.println(
"Map key:"&＃43;key.toString()&＃43;
" Map value:"&＃43;value.toString());
StringTokenizer itr &＃61;
new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
String tmp &＃61; itr.nextToken();
word.set(tmp);
context.write(word, one);
System.out.println(
"tmp:"&＃43;tmp&＃43;
" one:"&＃43;one);
}
System.out.println(
"context:"&＃43;context.toString());
}
}
注&＃xff1a;这里远哥要说一下“IntWritable one &＃61; new IntWritable(1);”的用意&＃xff0c;因为我们不管一个单词会出现几次&＃xff0c;只要出现&＃xff0c;我们就计算1次&＃xff0c;所以“context.write(word, one)”这行代码将一个单词写入的时候&＃xff0c;值永远是1&＃xff1b;
在运行的时候&＃xff0c;根据你文件中内容的情况&＃xff0c;上面的 map(Object key, Text value, Context context) 方法可能会被调用多次&＃xff0c;将本例子提供的文件内容执行后&＃xff0c;控制台输出内容如下(为了方便阅读&＃xff0c;我添加了一些换行)&＃xff1a;
TokenizerMapper.map...
Map key:0 Map value:lixy csy lixy zmde nitamade hehe
tmp:lixy one:1
tmp:csy one:1
tmp:lixy one:1
tmp:zmde one:1
tmp:nitamade one:1
tmp:hehe one:1
context:org.apache.hadoop.mapreduce.Mapper$Context&＃64;1af0b4a3
TokenizerMapper.map...
Map key:34 Map value:
context:org.apache.hadoop.mapreduce.Mapper$Context&＃64;1af0b4a3
TokenizerMapper.map...
Map key:36 Map value:realy amoeba woyou weibo hehe
tmp:realy one:1
tmp:amoeba one:1
tmp:woyou one:1
tmp:weibo one:1
tmp:hehe one:1
context:org.apache.hadoop.mapreduce.Mapper$Context&＃64;1af0b4a3
IntSumReducer.reduce...
val.get():1
Reduce key:amoeba Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;d5d4de6 Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:csy Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;d5d4de6 Result:1
IntSumReducer.reduce...
val.get():1
val.get():1
Reduce key:hehe Reduce result:2
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;d5d4de6 Result:2
IntSumReducer.reduce...
val.get():1
val.get():1
Reduce key:lixy Reduce result:2
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;d5d4de6 Result:2
IntSumReducer.reduce...
val.get():1
Reduce key:nitamade Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;d5d4de6 Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:realy Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;d5d4de6 Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:weibo Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;d5d4de6 Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:woyou Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;d5d4de6 Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:zmde Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;d5d4de6 Result:1
从TokenizerMapper的 map(Object key, Text value, Context context) 调用的信息输出情况可以分析出&＃xff0c;文件内容中有两行&＃xff0c;所以该方法一共调用了2次(因为TextInputFormat类型的&＃xff0c;都是按行处理)。
每一行的内容会在value参数中传进来&＃xff0c;也就是说每一行的内容都对应了一个key&＃xff0c;这个key为此行的开头位置在本文件中的所在位置(所以第1行的key是0&＃xff0c;第2行的key是34&＃xff0c;第3行的key是36)&＃xff0c;一般为数字的。
在这个map方法中&＃xff0c;我们可以加入一些自己的处理逻辑&＃xff0c;比如根据空格来取得每个单词&＃xff0c;然后我们需要将处理后的结果&＃xff0c;写入到context 参数中&＃xff0c;便于hadoop处理完后续的处理逻辑。(这里我们需要注意的是“IntWritable one”变量都是数值1)
上面看了map的过程&＃xff0c;接下来我们再看reduce的过程&＃xff0c;先看看 IntSumReducer 的代码&＃xff1a;
public class IntSumReducer extends Reducer {
private IntWritable result &＃61; new IntWritable();
public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
System.out.println("IntSumReducer.reduce...");
int sum &＃61; 0;
for (IntWritable val : values) {
sum &＃43;&＃61; val.get();
System.out.println("val.get():" &＃43; val.get());
}
result.set(sum);
context.write(key, result);
System.out.println("Reduce key:" &＃43; key.toString() &＃43; " Reduce result:" &＃43; result.get());
System.out.println("Reduce Context:" &＃43; context &＃43; " Result:" &＃43; result);
}
}
执行调用后&＃xff0c;控制台输出内容如下&＃xff1a;
IntSumReducer.reduce...
val.get():1
Reduce key:amoeba Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;6c04ab2f Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:csy Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;6c04ab2f Result:1
IntSumReducer.reduce...
val.get():2
Reduce key:hehe Reduce result:2
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;6c04ab2f Result:2
IntSumReducer.reduce...
val.get():2
Reduce key:lixy Reduce result:2
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;6c04ab2f Result:2
IntSumReducer.reduce...
val.get():1
Reduce key:nitamade Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;6c04ab2f Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:realy Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;6c04ab2f Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:weibo Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;6c04ab2f Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:woyou Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;6c04ab2f Result:1
IntSumReducer.reduce...
val.get():1
Reduce key:zmde Reduce result:1
Reduce Context:org.apache.hadoop.mapreduce.Reducer$Context&＃64;6c04ab2f Result:1
通过执行reduce(Text key, Iterable values, Context context) 方法&＃xff0c;奇迹发生了&＃xff0c;hadoop传到这里的参数&＃xff0c;已经去重了。什么意思呢&＃xff1f;就是说&＃xff0c;参数key里面是单词名称&＃xff0c;如果一个单词出现2次&＃xff0c;那么参数values里面就会2个值&＃xff0c;但是key只有1次。像“lixy”这个单词在第一行出现了2次&＃xff0c;那么这里的key只出现1次&＃xff0c;但是后面的values会有2个IntWritable&＃xff0c;并且值都是1&＃xff0c;这个为1的值其实就是你在map的时候&＃xff0c;自己定的。
这个例子只是简单的说明了MapReduce的一个简单使用&＃xff0c;实际上他的功能远不只这些&＃xff0c;还可以实现一些对数据库中数据的排序、统计等等&＃xff0c;特别是对一些非常非常庞大的数据表。当你有一个1T的文本内容&＃xff0c;需要统计里面每个单词分别出现多少次的时候&＃xff0c;一台计算机去计算&＃xff0c;会需要很长时间的&＃xff0c;有可能光加载就要很长时间&＃xff0c;但是如果你交给hadoop&＃xff0c;并且配了几台机器一起跑的话&＃xff0c;hadoop能把这1T的文本内容分成很多个小段&＃xff0c;分发到不同的物理机器上&＃xff0c;并行执行你的MapReduce逻辑&＃xff0c;然后将几台物理机器上处理完成的内容汇总后给你&＃xff0c;整个过程是分片、并行处理完成的&＃xff0c;效率大大提高。
京公网安备 11010802041100号