hadoop实现倒排索引

本文用hadoop实现倒排索引算法，用基本的分两步完成，不使用combine

第一步

读入文档，统计文档中各个单词的个数，与word count类似，但这里把word-filename组合起来作为一个key，并把中间结果写到磁盘中

InverseIndexStepTwo.java

package postlisting;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

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.StringUtils;

import java.io.IOException;

/**

 * 倒排索引步骤一，先做word count，不过现在的key是word-filename

 */

public class InverseIndexStepOne {

    public static class StepOneMapper extends Mapper<LongWritable, Text, Text, LongWritable>{

        @Override

        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String line = value.toString();

            // 切分出各个单词

            String[] fields = line.split(" ");

            // 获取文件切片

            FileSplit inputsplit = (FileSplit)context.getInputSplit();

            // 获取文件名

            String filename = inputsplit.getPath().getName();

            // 计数hello-->a.txt  1

            for(String field: fields){

                context.write(new Text(field+"-->"+filename), new LongWritable(1));

            }

        }

    }

    public static class  StepOneReducer extends Reducer<Text, LongWritable, Text, LongWritable>{

        @Override

        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

            long counter = 0;

            for (LongWritable value: values){

                counter += value.get();

            }

            context.write(key, new LongWritable(counter));

        }

    }

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

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);

        job.setJarByClass(InverseIndexStepOne.class);

        job.setMapperClass(StepOneMapper.class);

        job.setReducerClass(StepOneReducer.class);

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(LongWritable.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(LongWritable.class);

        // 检查输出文件夹是否已存在，如果存在先删除

        // 本地测试

        Path output = new Path("res/words/output/step1");

        FileSystem fs = FileSystem.get(conf);

        if(fs.exists(output)){

            fs.delete(output, true);

        }

        FileInputFormat.setInputPaths(job, new Path("res/words/input/"));

        FileOutputFormat.setOutputPath(job, output);

        System.out.println(job.waitForCompletion(true));

    }

}

输出结果

hello-->a.txt	2

hello-->b.txt	2

hello-->c.txt	2

jerry-->a.txt	1

jerry-->b.txt	3

jerry-->c.txt	1

tom-->a.txt	3

tom-->b.txt	1

tom-->c.txt	1

第二步

读取上一步的中间结果，解析并合并

InverseIndexStepOne.java

package postlisting;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

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.output.FileOutputFormat;

import java.io.IOException;

public class InverseIndexStepTwo {

    public static class StepTwoMapper extends Mapper<LongWritable, Text, Text, Text> {

        @Override

        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String line = value.toString();

            // hello-->a.txt    1

            String[] fields = line.split("\t");

            String[] wordAndFileName = fields[0].split("-->");

            String word = wordAndFileName[0];

            String fileName = wordAndFileName[1];

            long count = Long.parseLong(fields[1]);

            // <hello, a.txt-->3>

            context.write(new Text(word), new Text(fileName + "-->" + count));

        }

    }

    public static class StepTwoReducer extends Reducer<Text, Text, Text, Text>{

        @Override

        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            // 拿到的数据<hello, a.txt-->3, a.txt-->4,...>

            StringBuilder result = new StringBuilder();

            for (Text value:values){

                result.append(" ").append(value);

            }

            context.write(key, new Text(result.toString()));

        }

    }

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

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);

        job.setJarByClass(InverseIndexStepTwo.class);

        job.setMapperClass(StepTwoMapper.class);

        job.setReducerClass(StepTwoReducer.class);

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(Text.class);

        // 检查输出文件夹是否已存在，如果存在先删除

        Path output = new Path("res/words/output/step2");

        FileSystem fs = FileSystem.get(conf);

        if(fs.exists(output)){

            fs.delete(output, true);

        }

        FileInputFormat.setInputPaths(job, new Path("res/words/output/step1/"));

        FileOutputFormat.setOutputPath(job, output);

        System.out.println(job.waitForCompletion(true));

    }

}

输出结果

hello	 c.txt-->2 b.txt-->2 a.txt-->2

jerry	 c.txt-->1 b.txt-->3 a.txt-->1

tom	 c.txt-->1 b.txt-->1 a.txt-->3

小结

虽然用combine可以节省代码，但感觉分开写更加灵活，写个shell脚本组织一下就好，Map Reduce的强大之处也在与它的*组合。