MapReduce在倒排索引中練習

倒排索引是文件檢索系統中常用的數據結構，被廣泛應用於全文章搜索引擎。
 
通常情況下，倒排索引由一個單詞或詞組以及相關的文檔列表組成，文檔列表中的文檔或者是標識文檔的ID號，或者是指定文檔所在位置的URI；
 
在實際應用中，往往還需要給每個文檔加一個權值，用來指出每個文檔與搜索內容的相關度；
 
我的例子中，文檔內容如下：
 
Hadoop11:/home/in/win1 # hadoop fs -cat /user/root/in1/words.txt
 mapreduce is simple
 hadoop11:/home/in/win1 # hadoop fs -cat /user/root/in1/words1.txt
 mapreduce is powerfull and is simple
 hadoop11:/home/in/win1 # hadoop fs -cat /user/root/in1/words2.txt
 cat: File does not exist: /user/root/in1/words2.txt
 
我的目標結果：
 
and words1.txt:1;
 bye words3.txt:1;
 hello words3.txt:1;
 is words.txt:1;words1.txt:2;
 mapreduce words1.txt:1;words3.txt:2;words.txt:1;
 powerfull words1.txt:1;
 simple words1.txt:1;words.txt:1;
 
代碼清單，自己import吧：
 
 
 
public class indexSum {
  private static final Log log = LogFactory.getLog(LogCount.class);
 
 
 
 public static class TokenizerMapper1 extends
    Mapper<Object, Text, Text, Text> {
 
  private Text key1 = new Text();
  private Text value1 = new Text();
 
  public void map(Object key, Text value, Context context)
    throws IOException, InterruptedException {
 
  FileSplit split = (FileSplit) context.getInputSplit();
 
  StringTokenizer itr = new StringTokenizer(value.toString());
 
  while (itr.hasMoreTokens()) {
    key1.set(itr.nextToken() + ":"
      + split.getPath().getName().toString());
 
    value1.set("1");
 
    context.write(key1, value1);
    }
  }
  }
 
 public static class IntSumReducer extends Reducer<Text, Text, Text, Text> {
  private Text result = 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(":");
    result.set(key.toString().substring(splitIndex + 1) + ":" + sum);
    key.set(key.toString().substring(0, splitIndex));
    context.write(key, result);
  }
 
 }
 
 public static class IntSumReducer3 extends Reducer<Text, Text, Text, Text> {
  private Text result = new Text();
 
  public void reduce(Text key, Iterable<Text> values, Context context)
    throws IOException, InterruptedException {
 
  String valueArray = new String();
    for (Text value : values) {
    valueArray += value.toString() + ";";
    }
 
  result.set(valueArray);
 
  context.write(key, result);
  }
 
 }
 
 
 
 public static void main(String[] args) throws Exception {
  log.info("===============Begin at : " + new Date());
 
  String[] argsTemp = { "hdfs://hadoop11:8020/user/root/in1", "out4" };
  args = argsTemp;
 
  File jarFile = EJob.createTempJar("bin");
  Configuration conf = new Configuration();
  conf.set("hadoop.job.ugi", "root,root");
  conf.set("fs.default.name", "hdfs://hadoop11:8020/");
  conf.set("mapred.job.tracker", "hadoop11:8021");
 
  String[] otherArgs = new GenericOptionsParser(conf, args)
    .getRemainingArgs();
  if (otherArgs.length != 2) {
    System.err.println("Usage: wordcount <in> <out>");
    System.exit(2);
  }
  Job job = new Job(conf, "word count");
  job.setJarByClass(LogCount.class);
  ((JobConf) job.getConfiguration()).setJar(jarFile.toString());
 
 
  job.setMapperClass(TokenizerMapper1.class);
  job.setCombinerClass(IntSumReducer.class);
 
  job.setReducerClass(IntSumReducer3.class);
 
  FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
  FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
 
  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(Text.class);
 
  System.exit(job.waitForCompletion(true) ? 0 : 1);
 
  log.info("===============End at : " + new Date());
  }
 }