Example: WordCount v3.0

a
an
and
but
is
or
the
to
.
,

$ hadoop fs -put stop_words.text /user/cloudera/wordcount/

Code Changes

Lines 3-9: Import classes for handling file I/O and set support.

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.HashSet;
import java.util.Set;

Line 20: Import FileSplit, which is used to process a portion of an input file rather than the entire file at once.

import org.apache.hadoop.mapreduce.lib.input.FileSplit;

Line 25: Import StringUtils, which, not surprisingly, has a shed load of utilities for working with strings in Hadoop. See the StringUtils Javadoc to see all of the handy methods this class provides. This example uses the stringifyException() method to convert any IOExceptions to a string before passing it to the System.error.println() method.

import org.apache.hadoop.util.StringUtils;

Lines 40-48

for (int i = 0; i < args.length; i += 1) {
  if ("-skip".equals(args[i])) {
    job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
    i += 1;
    job.addCacheFile(new Path(args[i]).toUri());
    // this demonstrates logging
    LOG.info("Added file to the distributed cache: " + args[i]);
  }
}

Line 54: This time, you can add the optional combiner class to your job configuration. The combiner is run on each of the mapper machines in the cluster to process information locally before sending it to the reducer machine. The combiner can do anything you want in order to prepare the intermediate values prior to sending them to the reducer.

In this case, setting the Reduce class as the combiner counts up the keys locally. For example, of sending <elephant, 1> <elephant, 1> to the reducer, the mapper machine combines them together as <elephant, 2> before forwarding the results to the reducer. Combining values before transmission can save a great deal of bandwidth and transfer time in most cases.

job.setCombinerClass(Reduce.class);

Line 66: Add a variable to store an input chunk from a split file.

private String input;

Line 67: Create a Set of strings called patternsToSkip. This is the list of punctuation and superfluous words to be removed from the final results.

private Set<String> patternsToSkip = new HashSet<String>();

Lines 73-77: Convert information from split sources to a string for processing.

if (context.getInputSplit() instanceof FileSplit) {
  this.input = ((FileSplit) context.getInputSplit()).getPath().toString();
} else {
  this.input = context.getInputSplit().toString();
}

Lines 80-83: If the system variable wordcount.skip.patterns is true, get the list of patterns to skip from a distributed cache file and forward the URI of the file to the parseSkipFile method.

if (config.getBoolean("wordcount.skip.patterns", false)) {
  URI[] localPaths = context.getCacheFiles();
  parseSkipFile(localPaths[0]);
}

Lines 86-98: Get the distributed cache file from HDFS at the local URI. Read a line at a time until you run out of lines. Add each line to the set of strings to skip.

private void parseSkipFile(URI patternsURI) {
  LOG.info("Added file to the distributed cache: " + patternsURI);
  try {
    BufferedReader fis = new BufferedReader(new FileReader(new File(patternsURI.getPath()).getName()));
    String pattern;
    while ((pattern = fis.readLine()) != null) {
      patternsToSkip.add(pattern);
    }
  } catch (IOException ioe) {
    System.err.println("Caught exception while parsing the cached file '"
      + patternsURI + "' : " + StringUtils.stringifyException(ioe));
  }
}

Lines 108-110: Modify the if statement so that if the word variable is empty or it contains one of the defined patterns to skip, the for loop continues without writing a value to the context.

if (word.isEmpty() || patternsToSkip.contains(word)) {
  continue;
}