package com.IverseIndex; import java.io.IOException; 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.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class Entry { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration conf = new Configuration(); String[] otherarg = new GenericOptionsParser(conf, args) .getRemainingArgs(); if (otherarg.length != 2) { System.out.println("error!"); System.exit(2); } @SuppressWarnings("deprecation") Job job = new Job(conf, "InverseIndex"); job.setJarByClass(Entry.class); job.setMapperClass(InverseIndexMapper.class); job.setCombinerClass(InverseIndexCombiner.class); job.setReducerClass(InverseIndexReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(otherarg[0])); FileOutputFormat.setOutputPath(job, new Path(otherarg[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } } package com.IverseIndex; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.FileSplit; public class InverseIndexMapper extends Mapper<Object, Text, Text, Text> { private Text wordinfo = new Text(); private FileSplit split; public void map(Object key, Text value, Context context) throws IOException, InterruptedException { split = (FileSplit) context.getInputSplit(); StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { wordinfo.set(itr.nextToken() + ":" + split.getPath().toString()); context.write(wordinfo,new Text("1")); } } } package com.IverseIndex; import java.io.IOException; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class InverseIndexCombiner extends Reducer<Text, Text, Text, Text> { private Text wordinfo = new Text(); public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { int sum = 0; for (Text val : values) { sum +=Integer.parseInt(val.toString()); } int splitIndex = key.toString().indexOf(":"); wordinfo.set(key.toString().substring(splitIndex + 1) + ":" + String.valueOf(sum)); key.set(key.toString().substring(0, splitIndex)); context.write(key,wordinfo); } } package com.IverseIndex; import java.io.IOException; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class InverseIndexReducer extends Reducer<Text, Text, Text, Text> { public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { String file = new String(); for (Text value : values) { file += value.toString() + ";"; } context.write(key, new Text(file)); } }
倒排索引"是文档检索系统中最常用的数据结构,被广泛地应用于全文搜索引擎。它主要是用来存储某个单词(或词组)在一个文档或一组文档中的存储位置的映射,即提供了一种根据内容来查找文档的方式。由于不是根据文档来确定文档所包含的内容,而是进行相反的操作,因而称为倒排索引(Inverted Index)。
通常情况下,倒排索引由一个单词(或词组)以及相关的文档列表组成,文档列表中的文档或者是标识文档的ID号,或者是指文档所在位置的URL,
单词1出现在{文档1,文档4,文档13,……}中,单词2出现在{文档3,文档5,文档15,……}中,而单词3出现在{文档1,文档8,文档20,……}中。在实际应用中,还需要给每个文档添加一个权值,用来指出每个文档与搜索内容的相关度,如图所示。
最常用的是使用词频作为权重,即记录单词在文档中出现的次数。以英文为例,如图所示,索引文件中的"MapReduce"一行表示:"MapReduce"这个单词在文本T0中出现过1次,T1中出现过1次,T2中出现过2次。当搜索条件为"MapReduce"、"is"、"Simple"时,对应的集合为:{T0,T1,T2}∩{T0,T1}∩{T0,T1}={T0,T1},即文档T0和T1包含了所要索引的单词,而且只有T0是连续的。
更复杂的权重还可能要记录单词在多少个文档中出现过,以实现TF-IDF(Term Frequency-Inverse DocumentFrequency)算法,或者考虑单词在文档中的位置信息(单词是否出现在标题中,反映了单词在文档中的重要性)等。
样例输入如下所示。
1)file1:
MapReduceis simple
2)file2:
MapReduceis powerful is simple
3)file3:
HelloMapReduce bye MapReduce
样例输出如下所示。
MapReduce file1.txt:1;file2.txt:1;file3.txt:2;
is file1.txt:1;file2.txt:2;
simple file1.txt:1;file2.txt:1;
powerful file2.txt:1;
Hello file3.txt:1;
bye file3.txt:1;
实现"倒排索引"只要关注的信息为:单词、文档URL及词频。下面根据MapReduce的处理过程给出倒排索引的设计思路。
设计思路
1)Map过程:首先使用默认的TextInputFormat类对输入文件进行处理,得到文本中每行的偏移量及其内容。显然,Map过程首先必须分析输入的<key,value>对,得到倒排索引中需要的三个信息:单词、文档URL和词频,这里存在两个问题:第一,<key,value>对只能有两个值,在不使用Hadoop自定义数据类型的情况下,需要根据情况将其中两个值合并成一个值,作为key或value值;第二,通过一个Reduce过程无法同时完成词频统计和生成文档列表,所以必须增加一个Combine过程完成词频统计;
这里将单词和URL组成key值(如"MapReduce:file1.txt"),将词频作为value,这样做的好处是可以利用MapReduce框架自带的Map端排序,将同一文档的相同单词的词频组成列表,传递给Combine过程,实现类似于WordCount的功能。
2)Combine过程 :经过map方法处理后,Combine过程将key值相同的value值累加,得到一个单词在文档中的词频,如图所示。如果直接将图所示的输出作为Reduce过程的输入,在Shuffle过程时将面临一个问题:所有具有相同单词的记录(由单词、URL和词频组成)应该交由同一个Reducer处理,但当前的key值无法保证这一点,所以必须修改key值和value值。这次将单词作为key值,URL和词频组成value值(如"file1.txt:1")。这样做的好处是可以利用MapReduce框架默认的HashPartitioner类完成Shuffle过程,将相同单词的所有记录发送给同一个Reducer进行处理。
3)Reduce过程 :经过上述两个过程后,Reduce过程只需将相同key值的value值组合成倒排索引文件所需的格式即可,剩下的事情就可以直接交给MapReduce框架进行处理了。