Hadoop 倒排索引

　　倒排索引是文档检索系统中最常用的数据结构，被广泛地应用于全文搜索引擎。它主要是用来存储某个单词（或词组）在一个文档或一组文档中存储位置的映射，即提供了一种根据内容来查找文档的方式。由于不是根据文档来确定文档所包含的内容，而是进行相反的操作，因而称为倒排索引（Inverted Index）。

一、实例描述

　　倒排索引简单地就是，根据单词，返回它在哪个文件中出现过，而且频率是多少的结果。这就像百度里的搜索，你输入一个关键字，那么百度引擎就迅速的在它的服务器里找到有该关键字的文件，并根据频率和其他的一些策略（如页面点击投票率）等来给你返回结果。这个过程中，倒排索引就起到很关键的作用。

　　样例输入：

　　

　　样例输出：

　　

二、设计思路

　　倒排索引涉及几个过程：Map过程，Combine过程，Reduce过程。

　　Map过程： 

　　当你把需要处理的文档上传到hdfs时，首先默认的TextInputFormat类对输入的文件进行处理，得到文件中每一行的偏移量和这一行内容的键值对<偏移量，内容>做为map的输入。在改写map函数的时候，我们就需要考虑，怎么设计key和value的值来适合MapReduce框架，从而得到正确的结果。由于我们要得到单词,所属的文档URL,词频，而<key,value>只有两个值，那么就必须得合并其中得两个信息了。这里我们设计key=单词＋URL，value=词频。即map得输出为<单词＋URL，词频>，之所以将单词＋URL做为key，时利用MapReduce框架自带得Map端进行排序。

　　Combine过程：

　　Combine过程将key值相同得value值累加，得到一个单词在文档上得词频。但是为了把相同得key交给同一个reduce处理，我们需要设计为key=单词，value＝URL+词频。

　　Reduce过程：

　　Reduce过程其实就是一个合并的过程了，只需将相同的key值的value值合并成倒排索引需要的格式即可。

三、程序代码

　　程序代码如下：

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;


public class InvertedIndex {

    public static class Map extends Mapper<LongWritable, Text, Text, Text>{
        private static Text word = new Text();
        private static Text one = new Text();
        
        @Override
        protected void map(LongWritable key, Text value,Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            //  super.map(key, value, context);
            String fileName = ((FileSplit)context.getInputSplit()).getPath().getName();
            StringTokenizer st = new StringTokenizer(value.toString());
            while (st.hasMoreTokens()) {
                word.set(st.nextToken()+"\t"+fileName);
                context.write(word, one);
            }
        }
    }
    
    public static class Combine extends Reducer<Text, Text, Text, Text>{
        private static Text word = new Text();
        private static Text index = new Text();
        
        @Override
        protected void reduce(Text key, Iterable<Text> values,Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            //  super.reduce(arg0, arg1, arg2);
            String[] splits = key.toString().split("\t");
            if (splits.length != 2) {
                return ;
            }
            long count = 0;
            for(Text v:values){
                count++;
            }
            word.set(splits[0]);
            index.set(splits[1]+":"+count);
            context.write(word, index);
        }
    }
    
    public static class Reduce extends Reducer<Text, Text, Text, Text>{
        private static StringBuilder sub = new StringBuilder(256);
        private static Text index = new Text();
        
        @Override
        protected void reduce(Text word, Iterable<Text> values,Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            // super.reduce(arg0, arg1, arg2);
            for(Text v:values){
                sub.append(v.toString()).append(";");
            }
            index.set(sub.toString());
            context.write(word, index);
            sub.delete(0, sub.length());
        }
    }
    
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();
        if(otherArgs.length!=2){
            System.out.println("Usage:wordcount <in> <out>");
            System.exit(2);
        }
        Job job = new Job(conf,"Invert Index ");
        job.setJarByClass(InvertedIndex.class);
        
        job.setMapperClass(Map.class);
        job.setCombinerClass(Combine.class);
        job.setReducerClass(Reduce.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true)?0:1);
    }

}