Hadoop-Map/Reduce实现实现倒排索引

先来简单介绍一下什么是文档倒排索引
倒排索引是文档检索系统中最常见的数据结构，被广泛应用在全文搜索引擎上。主要用来存储某个单词（或词组）在一个文档或者一组文档中的存储位置的映射，即提供了一种根据内容来查找文档的方式。
简单点来讲呢，就是根据内容找文章。

倒排索引的概念说明白了，就该说说怎么用MapReduce实现。
测试数据奉上：
file1：MapReduce is simple
file2：MapReduce is powerful is simple
file3：Hello MapReduce bye MapReduce
输出的结果：
Hello file3.txt:1;
MapReduce file3.txt:2;file:2.txt:1;file1.txt:1;
bye file"3.txt:1;
is file2.txt:2;file1.txt:1;
powerful file2.txt:1;
simple file2.txt:1;file1.txt:1;

设计思路
map、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、
context.write("MapReduce:file1", 1) context.write("is:file1", 1) context.write("simple:file1", 1) context.write("MapReduce :file2", 1) context.write("is:file2", 1) context.write("powerful :file2", 1) context.write("is:file2", 1)

<"MapReduce:file1", {1}> <"is:file1", {1}> <"simple:file1", {1}> <"simple:file1", {1}> <"is:file2", {1, 1}> ..........................
combine、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、
context.write("MapReduce" , "file1:1") context.write("is" , "file1:1") context.write("simple" , "file1:1") context.write("MapReduce " , "file2:1") context.write("is" , "file2:2")................................
<"MapReduce",{ "file1:1","file2:1"}> <"is",{ "file1:1","file2:2"}> <"simple",{ "file1:1"}> .......................
reduce、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、
context.write("MapReduce","file1:1,file2:1")..................................

这个过程中的Combine是不可插拔的，也就是不可以省略的，因为它和Reduce的业务逻辑不一样。

代码奉上

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
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.output.FileOutputFormat;

public class InvertedIndex {

public static class InvertedMap extends
Mapper<LongWritable, Text, Text, IntWritable> {
private Text kText = new Text();
private IntWritable vIntWritable = new IntWritable(1);
private FileSplit split;

@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
String[] lineSplit = line.split("\t");
// 获取文档名称
split = (FileSplit) context.getInputSplit();
int indexOfFile = split.getPath().toString().indexOf("file");
String fileName = split.getPath().toString().substring(indexOfFile);

for (int i = 0; i < lineSplit.length; i++) {
kText.set(lineSplit[i] + "：" + fileName);
context.write(kText, vIntWritable);
}

}

}

public static class InvertedConbine extends
Reducer<Text, IntWritable, Text, Text> {
private Text kText = new Text();
private Text vText = new Text();

protected void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
// 词频统计
int sum = 0;
for (IntWritable v : values) {
sum += v.get();
}
int indexOf = key.toString().indexOf("：");
kText.set(key.toString().substring(0, indexOf));
vText.set(key.toString().substring(indexOf + 1) + "：" + sum);
context.write(kText, vText);

}

}

public static class InvertedReduce extends Reducer<Text, Text, Text, Text> {
private Text vText = new Text();

protected void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
String filelist = new String();
for (Text v : values) {
filelist += v.toString() + ";";
}
vText.set(filelist);
context.write(key, vText);
}

}

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);

job.setJarByClass(InvertedIndex.class);

job.setMapperClass(InvertedMap.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

job.setCombinerClass(InvertedConbine.class);

job.setReducerClass(InvertedReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

System.exit(job.waitForCompletion(true)? 0:1);
}

}