Hadoop（二）：MapReduce程序（Java）

 1 package hadoop;

 2 import org.apache.hadoop.conf.Configuration;　　// 读写和保存各种配置资源
 3 import org.apache.hadoop.fs.Path;　　// 保存文件或者目录的路径
 4 import org.apache.hadoop.io.IntWritable;　　// hadoop自身定义的整形类
 5 import org.apache.hadoop.io.Text;　　// hadoop自身定义的存储字符串的类
 6 import org.apache.hadoop.mapreduce.Job;　　// 每个hadoop任务是一个Job
 7 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;　　// 读取输入
 8 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;　　// 将结果存到输出文件
 9 import org.apache.hadoop.util.GenericOptionsParser;　　// 解析hadoop的命里行参数
10 
11 public class WordCount {
12     public static void main(String[] args) throws Exception {
13         Configuration conf = new Configuration();　　// 从hadoop配置文件里读取参数
14         String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();　　// 从hadoop命令行读取参数
15         if (otherArgs.length != 2) {　　// 从命令行读取的参数正常是两个，分别是输入文件和输出文件的目录
16             System.err.println("Usage: wordcount <in> <out>");
17             System.exit(2);
18         }
19         Job job = new Job(conf, "wordcount");　　// 定义一个新的Job，第一个参数是hadoop配置信息，第二个参数是Job的名字
20         job.setJarByClass(WordCount.class);　　// 根据WordCount类的位置设置Jar文件
21         job.setMapperClass(WordCountMapper.class);　　// 设置mapper文件　　
22         job.setReducerClass(WordCountReducer.class);　　// 设置reducer文件
23         job.setOutputKeyClass(Text.class);　　// 设定输出键的类型
24         job.setOutputValueClass(IntWritable.class);　　// 设定输出值的类型
25         FileInputFormat.addInputPath(job, new Path(otherArgs[0])); // 设定输入文件
26         FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); // 设定输出文件
27         System.exit(job.waitForCompletion(true) ? 0 : 1); // 开始执行Job
28     }
29 }

 1 package hadoop;
 2 
 3 import java.io.IOException;
 4 import java.util.StringTokenizer;　　// java提供的字符串分隔函数
 5 
 6 import org.apache.hadoop.io.IntWritable; 
 7 import org.apache.hadoop.io.Text;
 8 import org.apache.hadoop.mapreduce.Mapper;　　// hadoop提供的mapper基类，用户在此基础上进行自己的mapper程序开发
 9 
10 public class WordCountMapper extends Mapper<Object,Text,Text,IntWritable>{　　// ①
11     IntWritable one = new IntWritable(1);
12     Text word = new Text();
13 
14     public void map(Object key,Text value,Context context) throws IOException,InterruptedException{　　// ②
15         StringTokenizer itr = new StringTokenizer(value.toString());　　// 将字符串根据空格进行分割（value是Text类型的，所以需要将其转化成String类型进行处理）
16         while(itr.hasMoreTokens()){
17             word.set(itr.nextToken());
18             context.write(word,one);
19         }
20     }
21 }

 1 package hadoop;
 2 
 3 import java.io.IOException;
 4 
 5 import org.apache.hadoop.io.IntWritable;
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.mapreduce.Reducer; // Reducer基类　　
 8 
 9 public class WordCountReducer extends Reducer<Text,IntWritable,Text,IntWritable> {　　// ①
10     IntWritable result = new IntWritable();
11     public void reduce(Text key,Iterable<IntWritable>values,Context context) throws IOException,InterruptedException{ 　　// ②
12         int sum = 0;
13         for(IntWritable val:values){
14             sum += val.get();
15         }
16         result.set(sum);
17         context.write(key,result);
18     }
19 
20 }