package org.apache.hadoop.examples;

     import java.io.IOException;

     import java.util.Iterator;

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

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

         public DistinctWord() {

         }

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

             Configuration conf = new Configuration();

             //String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();

             String[] otherArgs = new String[]{"input","output"};  //设置输入和输出

             if(otherArgs.length < 2) {

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

                 System.exit(2);

             }

             Job job = Job.getInstance(conf, "distinct word");

             job.setJarByClass(DistinctWord.class);  //设置jar包所在路径

             //指定Mapper和Reducer类

             job.setMapperClass(DistinctWord.DistinctWordMapper.class);

             job.setCombinerClass(DistinctWord.DistinctWordReducer.class);

             job.setReducerClass(DistinctWord.DistinctWordReducer.class);

             //指定MapTask的输出类型

             job.setMapOutputKeyClass(Text.class);

             job.setMapOutputValueClass(NullWritable.class);

             //指定ReduceTask的输出类型

             job.setOutputKeyClass(Text.class);

             job.setOutputValueClass(NullWritable.class);

             //指定数据输入路径

             for(int i = 0; i < otherArgs.length - 1; ++i) {

                 FileInputFormat.addInputPath(job, new Path(otherArgs[i]));

             }

             //指定数据输出路径

             FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));

             //提交任务

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

         }

         //输出类型定义为NullWritable

         public static class DistinctWordMapper extends Mapper<Object, Text, Text, NullWritable> {

             private Text word = new Text();

             public DistinctWordMapper() {

             }

             public void map(Object key, Text value, Mapper<Object, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {

                 StringTokenizer itr = new StringTokenizer(value.toString());  //分词器

                 while(itr.hasMoreTokens()) {

                     this.word.set(itr.nextToken());

                     context.write(this.word, NullWritable.get());

                 }

             }

         }

         public static class DistinctWordReducer extends Reducer<Text, NullWritable, Text, NullWritable> {

             public DistinctWordReducer() {

             }

             //reduce方法每调用一次，就接收到一组相同的单词，所以直接输出一次key即可。

             public void reduce(Text key, Iterable<NullWritable> values, Reducer<Text, NullWritable, Text, NullWritable>.Context context) throws IOException, InterruptedException {

                 context.write(key, NullWritable.get());

             }

         }

     }