ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > Hadoop > 细节决定成败 MapReduce任务实战 倒排索引

细节决定成败 MapReduce任务实战 倒排索引

原创 Hadoop 作者：self_control 时间：2016-06-15 18:08:39 0 删除 编辑
今天在偶然看到一个博文，里面讲述如何使用mapreduce进行倒排索引处理。那就拿这个任务当成本篇实战任务吧。

一、任务描述


hdfs 上有三个文件，内容下上面左面框中所示。右框中为处理完成后的结果文件。
倒排索引（Inverted index），也常被称为反向索引、置入档案或反向档案，是一种索引方法，被用来存储在全文搜索下某个单词在一个文档或者一组文档中的存储位置的映射。它是文档检索系统中最常用的数据结构。通过倒排索引，可以根据单词快速获取包含这个单词的文档列表。倒排索引主要由两个部分组成：“单词词典”和“倒排文件”
这个任务与传统的倒排索引任务不同的地方是加上了每个文件中的频数。

二、实现思路

  • 首先关注结果中有文件名称，这个我们有两种方式处理：1、自定义InputFormat，在其中的自定义RecordReader中，直接通过InputSplit得到Path，继而得到FileName;2、在Mapper中，通过上下文可以取到Split，也可以得到fileName。这个任务中我们使用第二种方式，得到filename.
  • 在mapper中，得到filename 及 word，封装到一个自定义keu中。value 使用IntWritable。在map  中直接输出值为1的IntWritable对象。
  • 对进入reduce函数中的key进行分组控制，要求按word相同的进入同一次reduce调用。所以需要自定义GroupingComparator。

三、实现代码

自定义Key， WordKey代码。 注意这里面有个故意设置的坑。

点击(此处)折叠或打开

  1. package indexinverted;

  3. import java.io.DataInput;
  4. import java.io.DataOutput;
  5. import java.io.IOException;

  7. import org.apache.hadoop.io.WritableComparable;


  10. public class WordKey implements WritableComparable <WordKey> {

  12.         private String fileName;
  13.         private String word;

  15.         @Override
  16.         public void write(DataOutput out) throws IOException {
  17.                 out.writeUTF(fileName);
  18.                 out.writeUTF(word);
  19.         }

  21.         @Override
  22.         public void readFields(DataInput in) throws IOException {
  23.                 this.fileName = in.readUTF();
  24.                 this.word = in.readUTF();
  25.         }

  27.         @Override
  28.         public int compareTo(WordKey key) {
  29.                 int r = fileName.compareTo(key.fileName);
  30.                 if(r==0)
  31.                         r = word.compareTo(key.word);
  32.                 return r;
  33.         }

  35.         public String getFileName() {
  36.                 return fileName;
  37.         }

  39.         public void setFileName(String fileName) {
  40.                 this.fileName = fileName;
  41.         }

  43.         public String getWord() {
  44.                 return word;
  45.         }

  47.         public void setWord(String word) {
  48.                 this.word = word;
  49.         }
  50. }
Mapper、 Reducer、 IndexInvertedGroupingComparator （我喜欢把一些小的类当成内部类，放到Job类中，这样代码比较简单）
Reduce函数中处理输出结果有点繁琐，可以不用太关注。

点击(此处)折叠或打开

  1. package indexinverted;
  2. import java.io.IOException;
  3. import java.util.HashMap;
  4. import java.util.LinkedHashMap;

  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.conf.Configured;
  8. import org.apache.hadoop.fs.FileSystem;
  9. import org.apache.hadoop.fs.Path;
  10. import org.apache.hadoop.io.IntWritable;
  11. import org.apache.hadoop.io.LongWritable;
  12. import org.apache.hadoop.io.Text;
  13. import org.apache.hadoop.io.WritableComparable;
  14. import org.apache.hadoop.io.WritableComparator;
  15. import org.apache.hadoop.mapreduce.Job;
  16. import org.apache.hadoop.mapreduce.Mapper;
  17. import org.apache.hadoop.mapreduce.Reducer;
  18. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  19. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  20. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  21. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  22. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  23. import org.apache.hadoop.util.Tool;
  24. import org.apache.hadoop.util.ToolRunner;
  25. import org.apache.log4j.Logger;


  28. public class MyIndexInvertedJob extends Configured implements Tool{

  30.         public static class IndexInvertedMapper extends Mapper<LongWritable,Text,WordKey,IntWritable>{
  31.                 private WordKey newKey = new WordKey();
  32.                 private IntWritable ONE = new IntWritable(1);
  33.                 private String fileName ;
  34.                
  35.                 @Override
  36.                 protected void map(LongWritable key, Text value, Context context)
  37.                                 throws IOException, InterruptedException {
  38.                         newKey.setFileName(fileName);
  39.                         String words [] = value.toString().split(" ");
  40.                         for(String w:words){
  41.                                 newKey.setWord(w);
  42.                                 context.write(newKey, ONE);
  43.                         }
  44.                 }
  45.                 @Override
  46.                 protected void setup(Context context) throws IOException,
  47.                                 InterruptedException {
  48.                         FileSplit inputSplit = (FileSplit) context.getInputSplit();
  49.                         fileName = inputSplit.getPath().getName();
  50.                 }

  52.         }
  53.         public static class IndexInvertedReducer extends Reducer<WordKey,IntWritable,Text,Text>{
  54.                 private Text outputKey = new Text();

  56.                 @Override
  57.                 protected void reduce(WordKey key, Iterable<IntWritable> values,Context context)
  58.                                 throws IOException, InterruptedException {
  59.                         outputKey.set(key.getWord());
  60.                         LinkedHashMap <String,Integer> map = new LinkedHashMap<String,Integer>();
  61.                         for(IntWritable v :values){
  62.                                 if(map.containsKey(key.getFileName())){
  63.                                         map.put(key.getFileName(), map.get(key.getFileName())+ v.get());
  64.                                 }
  65.                                 else{
  66.                                         map.put(key.getFileName(), v.get());
  67.                                 }
  68.                         }
  69.                         StringBuilder sb = new StringBuilder();
  70.                         sb.append("{");
  71.                         for(String k: map.keySet()){
  72.                                 sb.append("(").append(k).append(",").append(map.get(k)).append(")").append(",");
  73.                         }
  74.                         sb.deleteCharAt(sb.length()-1).append("}");
  75.                         context.write(outputKey, new Text(sb.toString()));
  76.                 }

  78.         }
  79.         public static class IndexInvertedGroupingComparator extends WritableComparator{
  80.                 Logger log = Logger.getLogger(getClass());
  81.                 public IndexInvertedGroupingComparator(){
  82.                         super(WordKey.class,true);
  83.                 }

  85.                 @Override
  86.                 public int compare(WritableComparable a, WritableComparable b) {
  87.                         WordKey key1 = (WordKey) a;
  88.                         WordKey key2 = (WordKey) b;
  89.                         log.info("==============key1.getWord().compareTo(key2.getWord()):"+key1.getWord().compareTo(key2.getWord()));
  90.                         return key1.getWord().compareTo(key2.getWord());
  91.                 }

  93.         }
  94.         @Override
  95.         public int run(String[] args) throws Exception {
  96.                 Job job = Job.getInstance(getConf(), "IndexInvertedJob");
  97.                 job.setJarByClass(getClass());
  98.                 Configuration conf = job.getConfiguration();

  100.                 Path in = new Path("myinvertedindex/");
  101.                 Path out = new Path("myinvertedindex/output");
  102.                 FileSystem.get(conf).delete(out,true);
  103.                 FileInputFormat.setInputPaths(job, in);
  104.                 FileOutputFormat.setOutputPath(job, out);

  106.                 job.setInputFormatClass(TextInputFormat.class);
  107.                 job.setOutputFormatClass(TextOutputFormat.class);

  109.                 job.setMapperClass(IndexInvertedMapper.class);
  110.                 job.setMapOutputKeyClass(WordKey.class);
  111.                 job.setMapOutputValueClass(IntWritable.class);

  113.                 job.setReducerClass(IndexInvertedReducer.class);
  114.                 job.setOutputKeyClass(Text.class);
  115.                 job.setOutputValueClass(Text.class);

  117.                 job.setGroupingComparatorClass(IndexInvertedGroupingComparator.class);

  119.                 return job.waitForCompletion(true)?0:1;
  120.         }

  122.         public static void main(String [] args){
  123.                 int r = 0 ;
  124.                 try{
  125.                         r = ToolRunner.run(new Configuration(), new MyIndexInvertedJob(), args);
  126.                 }catch(Exception e){
  127.                         e.printStackTrace();
  128.                 }
  129.                 System.exit(r);
  130.         }

  132. }

四、查看结果

hadoop dfs -cat myinvertedindex/output/part-r-00000 

点击(此处)折叠或打开

  1. MapReduce {(1.txt,1)}
  2. is {(1.txt,1)}
  3. simple {(1.txt,1)}
  4. MapReduce {(2.txt,1)}
  5. is {(2.txt,2)}
  6. powerful {(2.txt,1)}
  7. simple {(2.txt,1)}
  8. Hello {(3.txt,1)}
  9. MapReduce {(3.txt,2)}
  10. bye {(3.txt,1)}
查看结果发现问题：单词并没有合并到一起，这会是什么原因？

五、深坑回填

查看结果发现问题：单词并没有合并到一起，这是什么原因？
GroupingComparator 是在什么样的基础上起作用的？是配置了按word相同输入到同一次reduce调用就一定会相同的word都进入同一个reduce调用吗？
NO！ Groupingcomparator中只是比较了相临的两个key是否相等。所有要结果正确，就要保存key的排序与GroupingComparator的排序是相协调。
问题出在了WordKey的排序是先按文件再按单词进行排，这样相临的key并不是单词相同的，而是文件相同的。
所以要改下WordKey的comparTo方法 。修复后代码如下：

点击(此处)折叠或打开

  1. package indexinverted;

  3. import java.io.DataInput;
  4. import java.io.DataOutput;
  5. import java.io.IOException;

  7. import org.apache.hadoop.io.WritableComparable;


  10. public class WordKey implements WritableComparable <WordKey> {

  12.         private String fileName;
  13.         private String word;

  15.         @Override
  16.         public void write(DataOutput out) throws IOException {
  17.                 out.writeUTF(fileName);
  18.                 out.writeUTF(word);
  19.         }

  21.         @Override
  22.         public void readFields(DataInput in) throws IOException {
  23.                 this.fileName = in.readUTF();
  24.                 this.word = in.readUTF();
  25.         }

  27.         @Override
  28.         public int compareTo(WordKey key) {
  29.                 int r = word.compareTo(key.word);
  30.                 if(r==0)
  31.                         r = fileName.compareTo(key.fileName);
  32.                 return r;
  33.         }

  35.         public String getFileName() {
  36.                 return fileName;
  37.         }

  39.         public void setFileName(String fileName) {
  40.                 this.fileName = fileName;
  41.         }

  43.         public String getWord() {
  44.                 return word;
  45.         }

  47.         public void setWord(String word) {
  48.                 this.word = word;
  49.         }
  50. }


运行结果

点击(此处)折叠或打开

  1. Hello {(3.txt,1)}
  2. MapReduce {(1.txt,1),(2.txt,1),(3.txt,2)}
  3. bye {(3.txt,1)}
  4. is {(1.txt,1),(2.txt,2)}
  5. powerful {(2.txt,1)}
  6. simple {(1.txt,1),(2.txt,1)}








来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/30066956/viewspace-2120238/ ，如需转载，请注明出处，否则将追究法律责任。

1
0
上一篇： 细节决定成败 MapReduce任务实战 Reduce Join
下一篇： HADOOP SPARK 集群环境搭建
请登录后发表评论 登录
全部评论
self_control

注册时间：2014-12-16

  • 博文量
    49
  • 访问量
    406895

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593