Hadoop学习系列之简单的带词频统计的倒排索引实现

2012st26

倒排索引概念

       倒排索引（Inverted index），也常被称为反向索引、置入档案或反向档案，是一种索引方法，被用来存储在全文搜索下某个单词在一个文档或者一组文档中的存储位置的映射。是文档检索中常用的数据结构。和索引的目的是一样的，倒排索引是为了提高检索效率而设计的。

       需要做倒排索引的文档通常都是数量巨大的，但是对不同的文档做索引可以并行，这事mapreduce最擅长做了，这就是我们今天所做的事。

设计说明：

       实验的输入是一组文档，输出的每行是：term      5@doc1       6@doc2       …

即每行输出一个单词，后面接着是该单词在某文档中的出现次数，每行中文档的是按序存放的，整个结果中单词也是排好序的。

       基于以上要求，各个类的设计如下：

       InvertedIndexMapper：

       由于读入的每行文本中包含大量的非单词字符，因此我们需要对此进行处理，此处的处理办法是将每个非单词字符([^0-9a-zA-Z])替换为空格，另外文档中大小写的问题也需要处理，我们在这儿统一将其转化为小写。另外在这一阶段需要做的事就是获取文件名，当然我们也可以自己定义一个InputFormat和RecordReader实现同样的效果，单是获取文件名我们可以采取这种相对简单的方式。剩下的事情就是提取单词进行发射：由于我们需要对出现某个单词的所有文件的文件名进行排序，可以利用系统自动对此进行排序，方法是将文件名和单词合并成这样的形式word@docid，最后我们在map阶段发射的键值对就是（word@docid， 1）。

       InvertedIndexCombiner：

       这个类所做的事就是对map发射的结果进行简单的合并，用wordcount里面的那个reduce类就可以。

       InvertedIndexPartitioner：

       在wordcount里面是没有这个类的，我们之所以要在这儿定制自己的partitioner，是因为我们发射过来的键值对是（word@docid, appeartime）形式的，也就是说word@doc1和word@doc2是不同的key，也就意味着可能会分配到不同的Reducer上。解决方案是将组合键<word@docid> 临时拆开，“蒙骗”partitioner按照<word>而不是<word@docid>进行分区选择正确的Reducer，这样可保证同一个word下的一组键值对一定被分区到同一个Reducer。

       InvertedIndexReducer：

       Reducer完成的工作就是对相同key的键值对进行合并，即将来的（word@doc1, appeartime1）、（word@doc1, appeartime2）……合并为（word   appeartime1@doc1  appeartime2@doc2  ……），采取的方案是使用一个vector和全局的oldword，处理逻辑如下：

       if(oldword == null || oldword != word){

                   if(oldword != null){

         　　　　　　发射(word, all_string_in_vector);

　　　　　　　}

　　　　　　vector清空;

　　　　　　oldword = word;

　　　}

       将appeartime@docid加入vector;

　　整个reduce工作完成后，还有最后一组索引没有发射出去，也就是说我们要自己完成清理工作，只需要重写Reducer类中的cleanup方法即可，此方法会在整个reduce完成之后被调用，我们在里面发射最后一组索引。

具体代码：

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
import org.apache.hadoop.util.GenericOptionsParser;

import com.sun.org.apache.bcel.internal.generic.NEW;
/*
 *代码版权：2012st26
 *by：backingwu 
 */
//map阶段输出为 key@DocID, value
public class InvertedIndex {
    
    private static Text oldkey = null;
    private static Vector<Text> vector1 = new Vector<Text>(); 
    
    public static class InvertedIndexMapper extends 
            Mapper<LongWritable, Text, Text, IntWritable>{
        private final static IntWritable one = new IntWritable(1);
        public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException{
            //获取文件名及预处理
            FileSplit fileSplit = (FileSplit)context.getInputSplit();
            String fileName = fileSplit.getPath().getName();
            String wordlineString = new String(value.toString().replaceAll("[^1-9a-zA-Z]", " "));
            StringTokenizer itr = new StringTokenizer(wordlineString.toLowerCase());
            
            while(itr.hasMoreTokens()){
                String tempKey = itr.nextToken();
                String temp2 = tempKey + "@" + fileName;
                context.write(new Text(temp2), one);
            }
        }
    }
    
    public static class InvertedIndexCombiner extends Reducer<Text, IntWritable, Text, IntWritable>{
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }
    
    public static class InvertedIndexPartioner extends HashPartitioner<Text, IntWritable>{

        public int getPartition(Text key, IntWritable value, int numReduceTasks) {
            // TODO Auto-generated method stub
            
            String term = key.toString().split("@")[0];
            super.getPartition(new Text(term), value, numReduceTasks);
            return 0;
        }
        
    }
    
    public static class InvertedIndexReducer extends Reducer<Text, IntWritable, Text, Text>{
        //reduce阶段输出为  key \tt1@doc1\tt2@doc2\tt3@doc3...
        public void reduce(Text key, Iterable<IntWritable> values, Context context
                )throws IOException, InterruptedException{
            String[] temp = key.toString().split("@");
            Text key1 = new Text(temp[0]);
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            Text valueText = new Text(String.valueOf(sum));
            if(oldkey == null || !oldkey.equals(key1) ){
                if(oldkey != null){
                    //发射
                    StringBuffer tmpString = new StringBuffer();
                    for(Text t: vector1){
                        tmpString.append(t.toString());
                    }
                    context.write(oldkey, new Text(tmpString.toString()));
                }
                
                vector1.clear();
                vector1.add(new Text(new String(" " + '\t')));
                oldkey = key1;
            }
            Text addText = new Text(valueText + "@" + temp[1] + '\t');
            vector1.add(addText);
        }
        
        
            
        //reduce阶段的清理工作
        protected void cleanup(Context context)
            throws IOException, InterruptedException{
            StringBuffer tmpString = new StringBuffer();
            for(Text t: vector1){
                tmpString.append(t.toString());
            }
            context.write(oldkey, new Text(tmpString.toString()));
        }
        
    }
            
    public static void main(String[] args){
        try{
            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, "invert index");
            job.setJarByClass(InvertedIndex.class);
            job.setInputFormatClass(TextInputFormat.class);
            job.setMapperClass(InvertedIndexMapper.class);
            job.setReducerClass(InvertedIndexReducer.class);
            job.setCombinerClass(InvertedIndexCombiner.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
            job.setPartitionerClass(InvertedIndexPartioner.class);
            FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
            FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
            FileSystem.get(conf).delete(new Path(otherArgs[1]));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        }catch(IOException e){
            e.printStackTrace();
        }catch(InterruptedException e){
            e.printStackTrace();
        }catch(ClassNotFoundException e){
            e.printStackTrace();
        }
        
    }
}

本文来自大笤帚---扫除一切障碍，奋勇向前冲的大笤帚！