WordCount源码分析

可以带着下面小问看回帖
1、Hadoop提供了哪些数据类型
2、作为Hadoop安装部署后，经典的Wordcount输出是怎样实现的

1 、特别数据类型介绍

Hadoop提供了如下内容的数据类型，这些数据类型都实现了WritableComparable接口，以便用这些类型定义的数据可以被序列化进行网络传输和文件存储，以及进行大小比较。

   BooleanWritable：标准布尔型数值

   ByteWritable：单字节数值

   DoubleWritable：双字节数

   FloatWritable：浮点数

   IntWritable：整型数

   LongWritable：长整型数

   Text：使用UTF8格式存储的文本

NullWritable：当<key,value>中的key或value为空时使用

2、WordCount分析

1）源代码程序

package org.apache.hadoop.examples;

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;

importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;

importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

importorg.apache.hadoop.util.GenericOptionsParser;

public class WordCount {

public static class  TokenizerMapper

extends Mapper<Object, Text, Text,IntWritable>{

private final static IntWritable one = newIntWritable(1);

private Text word = new Text();

public void map(Object key, Text value,Context context)

throws IOException, InterruptedException {

StringTokenizer itr = newStringTokenizer(value.toString());

while (itr.hasMoreTokens()) {

word.set(itr.nextToken());

context.write(word, one);

}

}

}

public static class  IntSumReducer

extendsReducer<Text,IntWritable,Text,IntWritable> {

private IntWritable result = newIntWritable();

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, result);

}

}

public static void  main (String[] args) throws Exception {

Configuration conf = new Configuration();

String[] otherArgs = newGenericOptionsParser(conf, args).getRemainingArgs();

if (otherArgs.length != 2) {

System.err.println("Usage: wordcount<in> <out>");

System.exit(2);

}

Job job = new Job(conf, "wordcount");

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, newPath(otherArgs[0]));

FileOutputFormat.setOutputPath(job, newPath(otherArgs[1]));

System.exit(job.waitForCompletion(true) ? 0: 1);

}

}

1）  Map过程

public static class  TokenizerMapper

extends Mapper<Object, Text, Text,IntWritable>{

private final static IntWritable one = newIntWritable(1);

private Text word = new Text();

public void map(Object key, Text value,Context context)

throws IOException, InterruptedException {

StringTokenizer itr = newStringTokenizer(value.toString());

while (itr.hasMoreTokens()) {

word.set(itr.nextToken());

context.write(word, one);

}

}

Map过程需要继承org.apache.hadoop.mapreduce包中 Mapper 类，并 重写 其map方法。通过在map方法中添加两句把key值和value值输出到控制台的代码，可以发现map方法中value值存储的是文本文件中的一行（以回车符为行结束标记），而key值为该行的首字母相对于文本文件的首地址的偏移量。然后StringTokenizer类将每一行拆分成为一个个的单词，并将<word,1>作为map方法的结果输出，其余的工作都交有 MapReduce框架 处理。

2）Reduce过程

public static class  IntSumReducer

extendsReducer<Text,IntWritable,Text,IntWritable> {

private IntWritable result = newIntWritable();

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, result);

}

}

Reduce过程需要继承org.apache.hadoop.mapreduce包中 Reducer 类，并 重写 其reduce方法。Map过程输出<key,values>中key为单个单词，而values是对应单词的计数值所组成的列表，Map的输出就是Reduce的输入，所以reduce方法只要遍历values并求和，即可得到某个单词的总次数。

3）执行MapReduce任务

public static void  main (String[] args) throws Exception {

Configuration conf = new Configuration();

String[] otherArgs = newGenericOptionsParser(conf, args).getRemainingArgs();

if (otherArgs.length != 2) {

System.err.println("Usage: wordcount<in> <out>");

System.exit(2);

}

Job job = new Job(conf, "wordcount");

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, newPath(otherArgs[0]));

FileOutputFormat.setOutputPath(job, newPath(otherArgs[1]));

System.exit(job.waitForCompletion(true) ? 0: 1);

}

在MapReduce中，由Job对象负责管理和运行一个计算任务，并通过Job的一些方法对任务的参数进行相关的设置。此处设置了使用TokenizerMapper完成Map过程中的处理和使用IntSumReducer完成Combine和Reduce过程中的处理。还设置了Map过程和Reduce过程的输出类型：key的类型为Text，value的类型为IntWritable。任务的输出和输入 路径 则由命令行参数指定，并由FileInputFormat和FileOutputFormat分别设定。完成相应任务的参数设定后，即可调用 job.waitForCompletion() 方法执行任务。

3、WordCount处理过程

1）将文件拆分成splits，由于测试用的文件较小，所以每个文件为一个split，并将文件按行分割形成<key,value>对，如图4-1所示。这一步由MapReduce框架自动完成，其中偏移量（即key值）包括了回车所占的字符数（Windows和Linux环境会不同）。

image

2）将分割好的<key,value>对交给用户定义的map方法进行处理，生成新的<key,value>对

image

3）得到map方法输出的<key,value>对后，Mapper会将它们按照key值进行排序，并执行Combine过程，将key至相同value值累加，得到Mapper的最终输出结果。

image

4）Reducer先对从Mapper接收的数据进行排序，再交由用户自定义的reduce方法进行处理，得到新的<key,value>对，并作为WordCount的输出结果，如图4-4所示。

5、MapReduce新旧改变

Hadoop最新版本的MapReduce Release 0.20.0的API包括了一个全新的MapreduceJAVA API，有时候也称为上下文对象。

新的API类型上不兼容以前的API，所以，以前的应用程序需要重写才能使新的API发挥其作用 。

新的API和旧的API之间有下面几个明显的区别。

新的API倾向于使用抽象类，而不是接口，因为这更容易扩展。例如，你可以添加一个方法(用默认的实现)到一个抽象类而不需修改类之前的实现方法。在新的API中，Mapper和Reducer是抽象类。

新的API是在org.apache.hadoop.mapreduce包(和子包)中的。之前版本的API则是放在org.apache.hadoop.mapred中的。

新的API广泛使用context object(上下文对象)，并允许用户代码与MapReduce系统进行通信。例如，MapContext基本上充当着JobConf的OutputCollector和Reporter的角色。

新的API同时支持"推"和"拉"式的迭代。在这两个新老API中，键/值记录对被推mapper中，但除此之外，新的API允许把记录从map()方法中拉出，这也适用于reducer。"拉"式的一个有用的例子是分批处理记录，而不是一个接一个。

新的API统一了配置。旧的API有一个特殊的JobConf对象用于作业配置，这是一个对于Hadoop通常的Configuration对象的扩展。在新的API中，这种区别没有了，所以作业配置通过Configuration来完成。作业控制的执行由Job类来负责，而不是JobClient，它在新的API中已经荡然无存。