MapReduce编程：单词去重

编程实现单词去重要用到NullWritable类型。

NullWritable：

NullWritable 是一种特殊的Writable 类型，由于它的序列化是零长度的，所以没有字节被写入流或从流中读出，可以用作占位符。比如，在MapReduce 中，在不需要这个位置的时候，键或值能够被声明为NullWritable，从而有效存储一个不变的空值。

通过调用NullWritable.get() 方法来检索。

单词去重我们最后要输出的形式是<单词>，所以值可以声明为NullWritable。

代码如下：

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  [...] ");
                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 {
            private Text word = new Text();

            public DistinctWordMapper() {
            }

            public void map(Object key, Text value, Mapper.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 {

            public DistinctWordReducer() {
            }

            //reduce方法每调用一次，就接收到一组相同的单词，所以直接输出一次key即可。
            public void reduce(Text key, Iterable values, Reducer.Context context) throws IOException, InterruptedException {
                context.write(key, NullWritable.get());
            }
        }


    }