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);
    }

}

?