hadoop倒排索引

 

1.前言

        學習hadoop的童鞋，倒排索引這個算法還是挺重要的。這是以后展開工作的基礎。首先，我們來認識下什么是倒拍索引：

        倒排索引簡單地就是：根據單詞，返回它在哪個文件中出現過，而且頻率是多少的結果。這就像百度里的搜索，你輸入一個關鍵字，那么百度引擎就迅速的在它的服務器里找到有該關鍵字的文件，並根據頻率和其他一些策略（如頁面點擊投票率）等來給你返回結果。這個過程中，倒排索引就起到很關鍵的作用。

2.分析設計

        倒排索引涉及幾個過程：Map過程，Combine過程，Reduce過程。下面我們來分析以上的過程。

2.1Map過程

        當你把需要處理的文檔上傳到hdfs時，首先默認的TextInputFormat類對輸入的文件進行處理，得到文件中每一行的偏移量和這一行內容的鍵值對<偏移量，內容>做為map的輸入。在改寫map函數的時候，我們就需要考慮，怎么設計key和value的值來適合MapReduce框架，從而得到正確的結果。由於我們要得到單詞,所屬的文檔URL,詞頻，而<key,value>只有兩個值，那么就必須得合並其中得兩個信息了。這里我們設計key=單詞＋URL，value=詞頻。即map得輸出為<單詞＋URL，詞頻>，之所以將單詞＋URL做為key，時利用MapReduce框架自帶得Map端進行排序。

        下面舉個簡單得例子： 

 

                                            圖1 map過程 輸入／輸出

2.2 Combine過程

 

      combine過程將key值相同得value值累加，得到一個單詞在文檔上得詞頻。但是為了把相同得key交給同一個reduce處理，我們需要設計為key=單詞，value＝URL+詞頻

  

                                             圖2 Combin過程 輸入/輸出 

2.3Reduce過程

      reduce過程其實就是一個合並的過程了，只需將相同的key值的value值合並成倒排索引需要的格式即可。 

 

                                          圖3 reduce過程 輸入/輸出

3.源代碼

 

package reverseIndex;

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.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 InvertedIndexMapper extends Mapper<Object, Text, Text, Text>{
        private Text keyInfo=new Text();
        private Text valueInfo=new Text();
        private FileSplit split;
        
        public void map(Object key,Text value,Context context)throws IOException,InterruptedException {
            //獲得<key,value>對所屬的對象
            split=(FileSplit)context.getInputSplit();
            StringTokenizer itr=new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                //key值有單詞和url組成，如"mapreduce:1.txt"
                keyInfo.set(itr.nextToken()+":"+split.getPath().toString());
                valueInfo.set("1");
                context.write(keyInfo, valueInfo);
            }
            
        }
    }
    public static class InvertedIndexCombiner extends Reducer<Text, Text, Text, Text>{
            private Text info=new Text();
            public void reduce(Text key,Iterable<Text> values,Context context)throws IOException,InterruptedException {
                //統計詞頻
                int sum=0;
                for (Text value:values) {
                    sum+=Integer.parseInt(value.toString());
                }
                
                int splitIndex=key.toString().indexOf(":");
                //重新設置value值由url和詞頻組成
                info.set(key.toString().substring(splitIndex+1)+":"+sum);
                //重新設置key值為單詞
                key.set(key.toString().substring(0,splitIndex));
                context.write(key, info);
            }
        }
    public static class InvertedIndexReduce extends Reducer<Text, Text, Text, Text> {
            private Text result=new Text();
            public void reduce(Text key,Iterable<Text>values,Context context) throws IOException,InterruptedException{
                //生成文檔列表
                String fileList=new String();
                for (Text value:values) {
                    fileList+=value.toString()+";";
                }
                result.set(fileList);
                context.write(key, result);
            }
        }
    
    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        Configuration conf=new Configuration();
        String[] otherArgs=new GenericOptionsParser(conf,args).getRemainingArgs();
        if (otherArgs.length!=2) {
            System.err.println("Usage:invertedindex<in><out>");
            System.exit(2);
        }
        Job job=new Job(conf,"InvertedIndex");
        job.setJarByClass(InvertedIndex.class);
        
        job.setMapperClass(InvertedIndexMapper.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        job.setCombinerClass(InvertedIndexCombiner.class);
        job.setReducerClass(InvertedIndexReduce.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        
        System.exit(job.waitForCompletion(true)?0:1);

    }

}