MapReduce算法一、简单求和计数（类似WordCount）

package MRDemo;

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.LongWritable;
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;

public class WordCount {
public static void main(String[] args) throws Exception {
//地址的个数为两个，
if(args.length!=2){
System.err.println("please input full path!");
System.exit(0);
}
Job job=new Job(new Configuration(), "WordCount");
//  将wordCount类打包运行
job.setJarByClass(WordCount.class);
//  设置输入路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
//  设置输出路径
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//  设置运行map函数的类
job.setMapperClass(WordCountMap.class);
//  设置运行reduce函数的类
job.setReducerClass(WordCountReduce.class);
//  设置map阶段输出的key、value的数据类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
//  设置reduce阶段输出的key、value的数据类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//  提交运行job作业
job.waitForCompletion(true);
}
/**
* KeyIn:每行文本偏移量的数据类型
* ValueIn：每行文本内容的数据类型
* KeyOut：Map输出的中间结果的key的数据类型
* ValueOut:Map输出的中间结果的value的数据类型
*
*
*/
public static class WordCountMap extends Mapper<LongWritable, Text, Text, IntWritable>{
protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,IntWritable>.Context context) throws java.io.IOException ,InterruptedException {
String[] lines  = value.toString().split(" ");
for (String info : lines) {
//        写成<hello , 1 > <world ,1>这样的形式
context.write(new Text(info), new IntWritable(1));

}
};
}

/**
* KeyIn:输入给Reduce处理的key的数据类型
* ValueIn：输入给Reduce处理的value的数据类型
* KeyOut:输出的最终结果的key的数据类型
* ValueOut：输出的最终结果的value的数据类型
*
*
*Map产生的中间结果，需要经过Shuffle阶段进行处理，
*将相同key值的value放到同一个values集合中，处理之后变成以下形式：
*<hello,{1,1}> <world,{1}>
*/
public static class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
protected void reduce(Text key, java.lang.Iterable<IntWritable> values, org.apache.hadoop.mapreduce
.Reducer<Text,IntWritable,Text,IntWritable>.Context context) throws java.io.IOException ,InterruptedException {
int sum = 0;
for (IntWritable count : values) {
sum =sum+count.get();//将HDFS格式通过get方法转化为java格式，

}
context.write(key, new IntWritable(sum));//通过new 的方式把java格式转为java类型
};
}
}