WordCount------自己写的第一个map/reduce程序------

//map:提取出整个文件中的有用内容，即---键值对<key,value>---
//map提取出的原始的键值对，经过合并，形成同一个key的所有value集<key,[value1,value2,value3,...]>----它是作为reduce的输入的
//    即，reduce的输入是一个特定类型的key,还有一个values集合。

package bin;

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 Count {

static class CountMapper extends Mapper<Object, Text,Text, IntWritable>{

private final static IntWritable one = new IntWritable(1);//声明整形成员变量one，初始化为1
private Text word=new Text();//声明string类型的成员变量
//定义正则表达式
private String pattern = "[^\\w]";//正则表达式，代表不是0-9, a-z, A-Z的所有其它字符

//定义map类继承mapped
public void map(Object key,Text value,Context context) {
String line = value.toString().replaceAll(pattern, " ");
StringTokenizer iStringTokenizer =new StringTokenizer(line);//该对象默认使用空格对字符串进行切分
while (iStringTokenizer.hasMoreElements()) {
word.set(iStringTokenizer.nextToken());
try {
context.write(word, one);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}

//定义reduce类
public static class CountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
//定义成员变量；
private IntWritable result = new IntWritable();
//reduce类中的reduce方法
public void reduce(Text key, Iterable<IntWritable> values, Context context) {
int sum = 0;
for (IntWritable value : values) {
sum +=value.get();//IntWritable的get()方法： Return the value of this IntWritable
}
result.set(sum);
try {
context.write(key, result);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//作业配置通过Configuration来完成配置作业的各种参数
Configuration configuration =new Configuration();
String[] otherArgs = new GenericOptionsParser(configuration,args).getRemainingArgs();
//GenericOptionsParser可以让Map/Reduce程序具有Hadoop常用的属性，对作业进行了部署。
//若每次传入的数据量不是两个，则显示报错信息。
if (otherArgs.length !=2) {
System.err.println("Usage: wordcount <in> <out>");
System.exit(2);
}
Job job =new Job(configuration, "tracert Word count");//新建一个job,给它起个名字，以便跟踪察看任务的执行情况--都有
job.setJarByClass(Count.class);//主类---都有
//当在hadoop集群上运行作业时，需要把代码打包成一个jar文件，hadoop会在集群分发这个文件，通过job的setJarByClass方法把代码所在的类设置好，
//hadoop会根据这个类找到所在的jar文件，这步之后，才可能会有hadoop去分发jar包.

//设置需要使用的map,combiner,reducer类
job.setMapperClass(CountMapper.class);
job.setCombinerClass(CountReducer.class);
job.setReducerClass(CountReducer.class);

//设置map reduce 的输出健和输出值类型
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);

}

}