Hadoop 中 MapReduce 快速入门

/*  
*WordCount的map程序  
*/

package com.lc.hadoop;

import java.io.IOException;  
import java.util.StringTokenizer;

//引入Hadoop本身提供的jar包  
import org.apache.hadoop.io.IntWritable;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.mapreduce.Mapper;

/*继承Mapper类，<Object,Text,Text,IntWritable>表示输入输出的key-value 类型*/  
public class TokenizerMapper extends Mapper<Object,Text,Text,IntWritable> {  
	IntWritable one=new IntWritable(1);  
	Text text=new Text();

	public void map(Object key,Text value,Context context)throws IOException,InterruptedException{  
/*key为输入的key，value为输入的value，因为用不上输入的key的类型，所以直接定义为Object类型，而Context是定义在Mapper类内部的，用于存储key-value键值对*/  
		StringTokenizer tokenizer=new StringTokenizer(value.toString());  
		while(tokenizer.hasMoreTokens()){  
			text.set(tokenizer.nextToken());  
			context.write(text,one);  
		}  
	}  
}  
```  
关于程序的几点解释：

- StringTokenizer类的作用是根据某一分隔符将String分隔开，默认是采用空格。  
- IntWritable 类表示的是一个整数，是一个以类表示的**可序列化的整数**  
- Text 类代表的是**可序列化的String类型**  
- Mapper 类将输入键值对映射到输出键值对，也就是 MapReduce 里的 Map 过程

经过map过程后，文章被分割成大量的k-v对，k为实际的单词，v均为1，下一步就是要将相同的单词合并在一起。

### Reduce函数  
Reduce函数的作用就是将相同的单词出现的次数合并在一起，代码如下：

```java  
/*  
*WordCount的reduce程序  
*/

package com.lc.hadoop;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.mapreduce.Reducer;

public class CountReducer extends Reducer<Text ,IntWritable,Text,IntWritable>{

	IntWritable result=new IntWritable();  
	public void reduce(Text key,Iterable<IntWritable> values,Context context)throws IOException,InterruptedException{  
		int sum=0;  
		for(IntWritable iw : values){  
			sum+=iw.get();  
		}  
		result.set(sum);  
		context.write(key,result);  
	}  
}  
```  
Reduce与Map函数有很多地方比较相似，均是继承了hadoop提供的jar包中的类，只是map函数继承了Mapper类，而reduce函数继承了Reducer类，输入输出的类型均是k-v键值对。而且reduce函数的输入就是map函数的输出。

### 主函数  
主函数的任务就是要创建一个任务，并且把map和reduce类都引进来，代码如下：

```java  
/*  
*WordCount的主程序  
*/

package com.lc.hadoop;

import java.io.IOException;

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.lib.input.FileInputFormat;  
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
import org.apache.hadoop.mapreduce.Job;  
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCount{

	public static void main(String[] args) throws Exception{  
		Configuration conf=new Configuration();//从hadoop配置文件中读取参数  
		//从命令行读取参数  
		String[] otherArgs=new GenericOptionsParser(conf,args).getRemainingArgs();

		if(otherArgs.length!=2){  
			System.out.println("Usage:wordcount <in> <out>");  
			System.exit(2);  
		}

		Job job=new Job(conf,"WordCount");  
		job.setJarByClass(WordCount.class);  
		job.setOutputKeyClass(Text.class);  
		job.setOutputValueClass(IntWritable.class);  
		job.setMapperClass(TokenizerMapper.class);  
		job.setReducerClass(CountReducer.class);  
		FileInputFormat.addInputPath(job,new Path(otherArgs[0]));  
		FileOutputFormat.setOutputPath(job,new Path(otherArgs[1]));  
		System.exit( (job.waitForCompletion(true)?0:1));

	}  
}