Salmon Run

So far, I had been using Hadoop 0.18 for my (admittedly rather minimal) Map-Reduce coding. Recently, when trying to find some random feature which I ended up never using, I discovered that it was only available in 0.19 upwards, and when I went to download, Hadoop was already up to 0.20, so I downloaded that instead. Now, Hadoop 0.20 came with some pretty massive API changes, which I did not know about when I downloaded it, and soon the code in my Eclipse IDE was awash in yellow squigglies and strikethroughs.

The yellow squigglies and strikethroughs are, of course, Eclipse telling me that I am looking at/using deprecated method calls. Hadoop 0.20 will support the old API as well, but it feels kind of wrong to write new code against a deprecated API knowing that it will disappear within a few releases. There is not much available by way of example, so I figured that if I could write my next Map-Reduce job using the new API, it would help me figure it out, so I did.

Algorithm

The job was to create a term-document matrix of the terms in about 150+ of my blog posts. I want to use the matrix for some experiments which I will describe later. The matrix is built in three steps as described in the table below:

Job-#	Phase	Input	Output
1	Map	<LineNum,Post>	List(<(docId,word),count>)
1	Reduce	<(docId,word),List(count)>	List(<(docId,word),count>)
2	Map	<(docId,word),count>	List(<word, count>)
2	Reduce (LongSumReducer)	<word,List(count)>	List(<word,count>)
3	Map	<(docId,word),count>	List(<docId,(word,count)>)
3	Reduce	<docId,List((word,count))>	List(<docId,CSV(count)>)

The input to this process is a sequence file of blogs. As always, the key is the line number (a LongWritable). In our case, the value is a MapWritable containing body, keyword and title. The output of the first phase is a list of (docId:word) => occurrence.

The second phase takes the output of the first phase, and just counts the word occurrences across the documents. This is used later to map words to their respective positions in the matrix.

The third stage uses the outputs of the first and second phases. The output of the first phase is converted to a List of docId => dense comma-separated list of positional occurrences. This can now be used as input to a matrix building component.

Code

Here is the code. The main method contains the driver code, and there are three Mapper and three Reducer classes built as public static inner classes. The new API seems to be cleaner than the old one, at least to me, but then I don't have a huge amount of unlearning to do, which may be the case for people who use it more heavily.

// Source: src/main/java/net/sf/jtmt/indexers/hadoop/TermDocumentMatrixGenerator.java
package net.sf.jtmt.indexers.hadoop;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;

import net.sf.jtmt.tokenizers.lucene.NumericTokenFilter;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
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.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.LowerCaseFilter;
import org.apache.lucene.analysis.PorterStemFilter;
import org.apache.lucene.analysis.StopFilter;
import org.apache.lucene.analysis.TokenStream;
import org.apache.lucene.analysis.standard.StandardFilter;
import org.apache.lucene.analysis.standard.StandardTokenizer;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.Field.Index;
import org.apache.lucene.document.Field.Store;
import org.apache.lucene.document.Field.TermVector;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.TermFreqVector;
import org.apache.lucene.index.IndexWriter.MaxFieldLength;
import org.apache.lucene.store.RAMDirectory;

/**
 * A 3-stage Hadoop job that reads a sequence file of blog articles,
 * and produces a term document vector of the article collection as
 * its output. The phases are as follows:
 * (1) Analyze each article using a custom Lucene analyzer and use
 *     Lucene's Term API to write out (docId:word) => occurrence.
 * (2) Read output of previous step and write out word => occurrence.
 * (3) Read output of step 2 to create a word to position map. Read
 *     output of step 1 to map docId => (word:occurrence) in the map
 *     phase, and docId => position,... in the reduce phase.
 * @author Sujit Pal
 * @version $Revision: 26 $
 */
public class TermDocumentMatrixGenerator { 

  /** Minimum occurrence for a given word in a document to be counted */
  private static final int MIN_OCCUR_THRESHOLD = 2;
  /** Key for the stop word file */
  private static final String STOPWORD_FILE_LOCATION_KEY = "stf.loc.key";
  /** Key for the terms directory generated in phase 2 */
  private static final String TERMS_DIR_KEY = "terms.dir.key";
  
  /* =============== Phase 1 ============== */
  
  /**
   * Map phase converts an input document into a set of (docId:word) 
   * => occurrence pairs. Stop words are removed, words are stemmed,
   * etc, using a custom analyzer. Words which occur below a threshold
   * are removed from consideration.
   */
  public static class Mapper1 extends 
      Mapper<Text,MapWritable,Text,LongWritable> {
    
    private Analyzer analyzer;
    
    protected void setup(Context context) 
        throws IOException, InterruptedException {
      Configuration conf = context.getConfiguration();
      FileSystem hdfs = FileSystem.get(conf);
      FSDataInputStream fis = hdfs.open(
        new Path(conf.get(STOPWORD_FILE_LOCATION_KEY)));
      final Set<String> stopset = new HashSet<String>();
      BufferedReader reader = new BufferedReader(new InputStreamReader(fis));
      String line = null;
      while ((line = reader.readLine()) != null) {
        stopset.add(line);
      }
      reader.close();
      fis.close();
      analyzer = new Analyzer() {
        public TokenStream tokenStream(String fieldName, Reader reader) {
          return new PorterStemFilter(
            new StopFilter(
              new LowerCaseFilter(
                new NumericTokenFilter(
                  new StandardFilter(
                    new StandardTokenizer(reader)))), stopset));
        }
      };
    }
    
    @Override
    public void map(Text key, MapWritable value, Context context) 
        throws IOException, InterruptedException {
      Text content = (Text) value.get(new Text("content"));
      TermFreqVector contentTf = getTermFrequencyVector(content.toString());
      String[] terms = contentTf.getTerms();
      int[] frequencies = contentTf.getTermFrequencies();
      for (int i = 0; i < terms.length; i++) {
        Text outputKey = new Text(StringUtils.join(
          new String[] {key.toString(), terms[i]}, ":"));
        context.write(outputKey, new LongWritable(frequencies[i]));
      }
    }

    private TermFreqVector getTermFrequencyVector(String content) 
        throws IOException {
      RAMDirectory ramdir = new RAMDirectory();
      IndexWriter writer = new IndexWriter(ramdir, analyzer, 
        MaxFieldLength.UNLIMITED);
      Document doc = new Document();
      doc.add(new Field("text", content.toString(), 
        Store.YES, Index.ANALYZED, TermVector.YES));
      writer.addDocument(doc);
      writer.commit();
      writer.close();
      IndexReader reader = IndexReader.open(ramdir);
      TermFreqVector termFreqVector = reader.getTermFreqVector(0, "text");
      reader.close();
      return termFreqVector;
    }
  }
  
  /**
   * Sums up the occurrences of (docId:word) occurrences and removes
   * those which occur infrequently.
   */
  public static class Reducer1 extends 
      Reducer<Text,LongWritable,Text,LongWritable> {
    
    @Override
    public void reduce(Text key, Iterable<LongWritable> values, 
        Context context) throws IOException, InterruptedException {
      long sum = 0L;
      for (LongWritable value : values) {
        sum += value.get();
      }
      if (sum > MIN_OCCUR_THRESHOLD) {
        context.write(key, new LongWritable(sum));
      }
    }
  }

  /* =============== Phase 2 ============== */

  /**
   * Extract the word from the key of the previous output, and emit
   * them. This will be used to build a word to position map.
   */
  public static class Mapper2 extends 
      Mapper<Text,LongWritable,Text,LongWritable> {
    
    private static final LongWritable ONE = new LongWritable(1);
    
    @Override
    public void map(Text key, LongWritable value, Context context) 
        throws IOException, InterruptedException {
      String word = StringUtils.split(key.toString(), ":")[1];
      context.write(new Text(word), ONE);
    }
  }
  
  /**
   * Aggregates the word count. 
   */
  public static class Reducer2 extends
      Reducer<Text,LongWritable,Text,LongWritable> {

    @Override
    public void reduce(Text key, Iterable<LongWritable> values, 
        Context context) throws IOException, InterruptedException {
      long occurrences = 0;
      for (LongWritable value : values) {
        occurrences++;
      }
      context.write(key, new LongWritable(occurrences));
    }
  }
  
  /* =============== Phase 3 ============== */

  /**
   * Reads the output of step 1 and emits docId => (word:occurrence)
   * pairs.
   */
  public static class Mapper3 extends Mapper<Text,LongWritable,Text,Text> {

    @Override
    public void map(Text key, LongWritable value, Context context) 
        throws IOException, InterruptedException {
      String[] pair = StringUtils.split(key.toString(), ":");
      context.write(new Text(pair[0]), 
        new Text(StringUtils.join(
        new String[] {pair[1], String.valueOf(value.get())}, ":")));
    }
  }

  /**
   * Converts the output of step 2 into a map of word and position.
   * Flattens the docId => (word:occurrence) pairs to docId => dense
   * positional list of occurrences. 
   */
  public static class Reducer3 extends Reducer<Text,Text,Text,Text> {

    private Map<String,Integer> terms = new HashMap<String,Integer>();
    
    @Override
    public void setup(Context context) 
        throws IOException, InterruptedException {
      Configuration conf = context.getConfiguration();
      FileSystem hdfs = FileSystem.get(conf);
      FileStatus[] partFiles = hdfs.listStatus(
        new Path(conf.get(TERMS_DIR_KEY)));
      for (FileStatus partFile : partFiles) {
        if (! partFile.getPath().getName().startsWith("part-r")) {
          continue;
        }
        FSDataInputStream fis = hdfs.open(partFile.getPath());
        BufferedReader reader = new BufferedReader(
          new InputStreamReader(fis));
        String line = null;
        int i = 0;
        while ((line = reader.readLine()) != null) {
          String term = StringUtils.split(line, "\t")[0];
          terms.put(term, i);
          i++;
        }
        reader.close();
        fis.close();
      }
    }

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context)
        throws IOException, InterruptedException {
      SortedMap<String,Long> frequencies = new TreeMap<String,Long>();
      for (Text value : values) {
        String[] parts = StringUtils.split(value.toString(), ":");
        String word = parts[0];
        Long occurrence = new Long(parts[1]);
        frequencies.put(word, occurrence);
      }
      context.write(key, flatten(frequencies));
    }

    private Text flatten(SortedMap<String,Long> frequencies) {
      long[] array = new long[terms.size()];
      for (String word : frequencies.keySet()) {
        int pos = terms.get(word);
        array[pos] = frequencies.get(word);
      }
      StringBuilder buf = new StringBuilder();
      for (int i = 0; i < array.length; i++) {
        if (i > 0) {
          buf.append(",");
        }
        buf.append(String.valueOf(array[i]));
      }
      return new Text(buf.toString());
    }
  }
    
  /**
   * Calling method.
   * @param argv command line args.
   * @throws Exception if thrown.
   */
  public static void main(String[] argv) throws Exception {
    Configuration conf = new Configuration();
    String[] otherArgs = new GenericOptionsParser(
      conf, argv).getRemainingArgs();
    if (otherArgs.length != 4) {
      System.err.println("Usage: tdmg <prefix> <stopwords> <indir> <outdir>");
      System.exit(-1);
    }
    Path basedir = new Path(otherArgs[0] + otherArgs[2]).getParent();
    /* phase 1: convert doc file to (docId:word) => occurrence */
    Job job1 = new Job(conf, "phase-1");
    job1.getConfiguration().set(STOPWORD_FILE_LOCATION_KEY, 
      otherArgs[0] + otherArgs[1]);
    job1.getConfiguration().set("foo", "bar");
    FileInputFormat.addInputPath(job1, 
      new Path(otherArgs[0] + otherArgs[2]));
    FileOutputFormat.setOutputPath(job1, new Path(basedir, "temp1"));
    job1.setJarByClass(TermDocumentMatrixGenerator.class);
    job1.setMapperClass(Mapper1.class);
    job1.setReducerClass(Reducer1.class);
    job1.setInputFormatClass(SequenceFileInputFormat.class);
    job1.setOutputFormatClass(SequenceFileOutputFormat.class);
    job1.setMapOutputKeyClass(Text.class);
    job1.setMapOutputValueClass(LongWritable.class);
    job1.setOutputKeyClass(Text.class);
    job1.setOutputValueClass(LongWritable.class);
    job1.setNumReduceTasks(2);
    boolean job1Success = job1.waitForCompletion(true);
    if (! job1Success) {
      System.err.println("Job1 failed, exiting");
      System.exit(-1);
    }
    /* phase 2: write out unique terms with indexes to terms directory */
    Job job2 = new Job(conf, "phase-2");
    FileInputFormat.addInputPath(job2, new Path(basedir, "temp1"));
    FileOutputFormat.setOutputPath(job2, new Path(basedir, "temp2"));
    job2.setJarByClass(TermDocumentMatrixGenerator.class);
    job2.setMapperClass(Mapper2.class);
    job2.setReducerClass(Reducer2.class);
    job2.setInputFormatClass(SequenceFileInputFormat.class);
    job2.setOutputFormatClass(TextOutputFormat.class);
    job2.setMapOutputKeyClass(Text.class);
    job2.setMapOutputValueClass(LongWritable.class);
    job2.setOutputKeyClass(Text.class);
    job2.setOutputValueClass(LongWritable.class);
    job2.setNumReduceTasks(2);
    boolean job2Success = job2.waitForCompletion(true);
    if (! job2Success) {
      System.err.println("Job2 failed, exiting");
      System.exit(-1);
    }
    /* phase 3: flatten to docId => occurrence,... */
    Job job3 = new Job(conf, "phase-3");
    job3.getConfiguration().set(TERMS_DIR_KEY, basedir + "/temp2"); 
    FileInputFormat.addInputPath(job3, new Path(basedir, "temp1"));
    FileOutputFormat.setOutputPath(job3, 
      new Path(otherArgs[0] + otherArgs[3]));
    job3.setJarByClass(TermDocumentMatrixGenerator.class);
    job3.setMapperClass(Mapper3.class);
    job3.setReducerClass(Reducer3.class);
    job3.setInputFormatClass(SequenceFileInputFormat.class);
    job3.setOutputFormatClass(TextOutputFormat.class);
    job3.setMapOutputKeyClass(Text.class);
    job3.setMapOutputValueClass(Text.class);
    job3.setOutputKeyClass(Text.class);
    job3.setOutputValueClass(Text.class);
    job3.setNumReduceTasks(2);
    System.exit(job3.waitForCompletion(true) ? 0 : 1);
  }
}

I ran this in local mode and pseudo-distributed mode. I had to change the code a bit for the pseudo-distributed mode (I think mostly its the file names), but here is the script I used to run it.

#!/bin/bash
# Source: ./src/main/scripts/hadoop/run_tdmg.sh
# Script to call Term Document Matrix Generator

## CONFIG ##
M2_REPO=/home/sujit/.m2/repository
HADOOP_HOME=/opt/hadoop-0.20.0
PROJECT_BASEDIR=/home/sujit/src/jtmt
MODE=p # mode can be (l)ocal or (p)seudo-distributed
if [ $MODE == "l" ]; then
  PROTOCOL_PREFIX=""
  STOPWORD_FILE=$PROJECT_BASEDIR/src/main/resources/stopwords.txt
  INPUT_DIR=$PROJECT_BASEDIR/src/test/resources/hac/inputs
  OUTPUT_DIR=$PROJECT_BASEDIR/src/test/resources/hac/outputs
  TEMP_DIRS=$PROJECT_BASEDIR/src/test/resources/hac/temp*
else
  PROTOCOL_PREFIX=hdfs://localhost:54310
  STOPWORD_FILE=/user/sujit/tdmg/resources/stopwords.txt
  INPUT_DIR=/user/sujit/tdmg/inputs
  OUTPUT_DIR=/user/sujit/tdmg/outputs
  TEMP_DIRS=/user/sujit/tdmg/temp*
fi
## CONFIG ##

# for local mode
if [ $MODE == "l" ]; then
  export HADOOP_CLASSPATH=$CLASSPATH:\
$M2_REPO/org/apache/lucene/lucene-core/2.4.0/lucene-core-2.4.0.jar:\
$M2_REPO/org/apache/lucene-analyzers/2.3.0/lucene-analyzers-2.3.0.jar:\
$M2_REPO/commons-lang/commons-lang/2.1/commons-lang-2.1.jar:\
$PROJECT_BASEDIR/target/jtmt-1.0-SNAPSHOT.jar
fi

cd $HADOOP_HOME
if [ $MODE == "l" ]; then
  rm -rf $OUTPUT_DIR
  rm -rf $TEMP_DIRS
  # no special packaging required for local mode
  bin/hadoop jar $PROJECT_BASEDIR/target/jtmt-1.0-SNAPSHOT.jar net.sf.jtmt.indexers.hadoop.TermDocumentMatrixGenerator "" $STOPWORD_FILE $INPUT_DIR $OUTPUT_DIR
else
  bin/hadoop fs -rmr $OUTPUT_DIR
  bin/hadoop fs -rmr $TEMP_DIRS
  bin/hadoop jar $PROJECT_BASEDIR/target/tdm-generator.jar $PROTOCOL_PREFIX $STOPWORD_FILE $INPUT_DIR $OUTPUT_DIR
fi
cd -
unset HADOOP_CLASSPATH

Similarities and Differences

The main difference of course is that you extend mapreduce.Mapper and mapreduce.Reducer instead of extending MapReduceBase and implementing either mapred.Mapper or mapred.Reducer. The Context provides a more unified way to pass stuff around, although I had some trouble (which I later fixed) passing parameters to the Mapper and Reducer classes. Another difference seems to be that the ToolRunner is no longer needed.

I did miss the "convenience" Mappers and Reducers (such as LongSumReducer), but I am guessing that they will make their way back in once the old API goes away.

Among the similarities, you still have to package your application code and third party dependencies in the same way as before, something I was hoping would go away once I learned about HADOOP_CLASSPATH, but I guess it makes sense to keep it the way it is.

Salmon Run

Friday, September 18, 2009

Generating a TD Matrix using Hadoop 0.20

Algorithm

Code

Similarities and Differences

Posts

Labels

Blogs I Read

About me

My Nerd Rating

Visitor Map

Contact Me