mapreduce (五) MapReduce实现倒排索引 修改版 combiner是把同一个机器上的多个map的结果先聚合一次

（总感觉上一篇的实现有问题）http://www.cnblogs.com/i80386/p/3444726.html   combiner是把同一个机器上的多个map的结果先聚合一次
现重新实现一个：

思路：
第一个mapreduce仅仅做  <word_docid,count>的统计，即某个单词在某一篇文章里出现的次数。（原理跟wordcount一样，只是word变成了word_docid)
第二个mapreduce将word_docid在map阶段拆开，重新组合为<word,docid_count> 然后在combine和reduce阶段（combine和reduce是同一个函数）组合为 <word,doc1:count1,doc2:count2,doc3:count3>这种格式import java.io.IOException;

1 思路：
0.txt MapReduce is simple
1.txt MapReduce is powerfull is simple
2.txt Hello MapReduce bye MapReduce

采用两个JOB的形式实现
一：第一个JOB（跟wordcount一致，只是wordcount中的word换做了word:dicid）
1 map函数：context.write(word:docid, 1) 即将word:docid作为map函数的输出
输出key        输出value
MapReduce:0.txt 1
is:0.txt 1
simple:0.txt 1
Mapreduce:1.txt 1
is:1.txt 1
powerfull:1.txt 1
is:1.txt 1
simple:1.txt 1
Hello:2.txt 1
MapReduce:2.txt 1
bye:2.txt 1
MapReduce:2.txt 1
2 Partitioner函数：HashPartitioner
略，根据map函数的输出key（word：docid）进行分区
3 reduce函数：累加输入values
输出key    输出value
MapReduce:0.txt 1 => MapReduce 0.txt:1 
is:0.txt 1        => is 0.txt:1
simple:0.txt 1    => simple 0.txt:1
Mapreduce:1.txt 1 => Mapreduce 1.txt:1
is:1.txt 2        => is 1.txt:2
powerfull:1.txt 1 => powerfull 1.txt:1
simple:1.txt 1    => simple 1.txt:1
Hello:2.txt 1     => Hello 2.txt:1
MapReduce:2.txt 2 => MapReduce 2.txt:2
bye:2.txt 1       => bye 2.txt:1
二：第二个JOB
1 map函数：
输入key    输入value  输出key    输出value
MapReduce:0.txt 1 => MapReduce 0.txt:1 
is:0.txt 1        => is 0.txt:1
simple:0.txt 1    => simple 0.txt:1
Mapreduce:1.txt 1 => Mapreduce 1.txt:1
is:1.txt 2        => is 1.txt:2
powerfull:1.txt 1 => powerfull 1.txt:1
simple:1.txt 1    => simple 1.txt:1
Hello:2.txt 1     => Hello 2.txt:1
MapReduce:2.txt 2 => MapReduce 2
2 reduce函数 （组合values）
输出key    输出value
MapReduce 0.txt:1，1.txt:1 2.txt:2
is 0.txt:1，is 1.txt:2
simple 0.txt:1，1.txt:1
powerfull 1.txt:1
Hello 2.txt:1
bye 2.txt:1

import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
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.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
import org.apache.hadoop.mapreduce.lib.reduce.IntSumReducer;

public class MyInvertIndex {

    public static class SplitMapper extends
            Mapper<Object, Text, Text, IntWritable> {

        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            FileSplit split = (FileSplit) context.getInputSplit();
            //String pth = split.getPath().toString();
            String name = split.getPath().getName();
            String[] tokens = value.toString().split("\s");
            for (String token : tokens) {
                context.write(new Text(token + ":" + name), new IntWritable(1));
            }
        }
    }

    public static class CombineMapper extends
            Mapper<Text, IntWritable, Text, Text> {

        public void map(Text key, IntWritable value, Context context)
                throws IOException, InterruptedException {
            int splitIndex = key.toString().indexOf(":");
            context.write(new Text(key.toString().substring(0, splitIndex)),
                    new Text(key.toString().substring(splitIndex + 1) + ":"
                            + value.toString()));
        }
    }

    public static class CombineReducer extends Reducer<Text, Text, Text, Text> {
        public void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            StringBuffer buff = new StringBuffer();
            for (Text val : values) {
                buff.append(val.toString() + ",");
            }
            context.write(key, new Text(buff.toString()));
        }
    }

    public static void main(String[] args) throws IOException,
            ClassNotFoundException, InterruptedException {

        String dir_in = "hdfs://localhost:9000/in_invertedindex";
        String dir_out = "hdfs://localhost:9000/out_invertedindex";

        Path in = new Path(dir_in);
        Path out = new Path(dir_out);
        Path path_tmp = new Path("word_docid"
                + Integer.toString(new Random().nextInt(Integer.MAX_VALUE)));

        Configuration conf = new Configuration();

        try {
            Job countJob = new Job(conf, "invertedindex_count");

            countJob.setJarByClass(MyInvertIndex.class);

            countJob.setInputFormatClass(TextInputFormat.class);
            countJob.setMapperClass(SplitMapper.class);
            countJob.setCombinerClass(IntSumReducer.class);
            countJob.setPartitionerClass(HashPartitioner.class);
            countJob.setMapOutputKeyClass(Text.class);
            countJob.setMapOutputValueClass(IntWritable.class);

            FileInputFormat.addInputPath(countJob, in);

            countJob.setReducerClass(IntSumReducer.class);
            // countJob.setNumReduceTasks(1);
            countJob.setOutputKeyClass(Text.class);
            countJob.setOutputValueClass(IntWritable.class);
            countJob.setOutputFormatClass(SequenceFileOutputFormat.class);

            FileOutputFormat.setOutputPath(countJob, path_tmp);

            countJob.waitForCompletion(true);

            Job combineJob = new Job(conf, "invertedindex_combine");

            combineJob.setJarByClass(MyInvertIndex.class);

            combineJob.setInputFormatClass(SequenceFileInputFormat.class);
            combineJob.setMapperClass(CombineMapper.class);
            combineJob.setCombinerClass(CombineReducer.class);
            combineJob.setPartitionerClass(HashPartitioner.class);
            combineJob.setMapOutputKeyClass(Text.class);
            combineJob.setMapOutputValueClass(Text.class);

            FileInputFormat.addInputPath(combineJob, path_tmp);

            combineJob.setReducerClass(CombineReducer.class);
            // combineJob.setNumReduceTasks(1);
            combineJob.setOutputKeyClass(Text.class);
            combineJob.setOutputValueClass(Text.class);
            combineJob.setOutputFormatClass(TextOutputFormat.class);

            FileOutputFormat.setOutputPath(combineJob, out);

            combineJob.waitForCompletion(true);

        } finally {
            FileSystem.get(conf).delete(path_tmp, true);
        }

    }
}

运行结果：
Hello    2.txt:1,,
MapReduce    2.txt:2,1.txt:1,0.txt:1,,
bye    2.txt:1,,
is    1.txt:2,0.txt:1,,
powerfull    1.txt:1,,
simple    1.txt:1,0.txt:1,,