source: sample/hadoop-0.17/tw/org/nchc/util/SequenceFileProcessor.java @ 20

/**
 * Program: BuildHTable.java
 * Editor: Waue Chen 
 * From :  NCHC. Taiwn
 * Last Update Date: 07/02/2008
 * Upgrade to 0.17
 * Re-code from : Cloud9: A MapReduce Library for Hadoop
 */


package tw.org.nchc.util;

import java.io.IOException;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.JobConf;

/**
 * Upgrade from hadoop 0.16 to 0.17 
 * <p>
 * Harness for processing one or more {@link SequenceFile}s within a single
 * process. This class is useful when you want to iterate through all key-value
 * pairs in a SequenceFile outside the context of a MapReduce task (or where
 * writing the computation as a MapReduce would be overkill). One example usage
 * case is to sum up all the values in a SequenceFile &mdash; this may be useful
 * if you want to make sure probabilities sum to one. Here's the code fragment
 * that would accomplish this:
 * </p>
 * 
 * <pre>
 * KeyValueProcess&lt;Tuple, FloatWritable&gt; process = SequenceFileProcessor
 *    .&lt;Tuple, FloatWritable&gt; process(&quot;foo&quot;,
 *        new KeyValueProcess&lt;Tuple, FloatWritable&gt;() {
 *          public float sum = 0.0f;
 * 
 *          public void process(Tuple tuple, FloatWritable f) {
 *            sum += f.get();
 *          }
 * 
 *          public void report() {
 *            setProperty(&quot;sum&quot;, sum);
 *          }
 *        });
 * 
 * float sum = (Float) process.getProperty(&quot;sum&quot;);
 * </pre>
 * 
 * <p>
 * The static method takes a path and and a {@link KeyValueProcess}. This
 * example uses an anonymous inner class to make the code more concise; the
 * static method returns the <code>KeyValueProcess</code> so that you can
 * retrieve results from it. The path can either be a file or a directory; if it
 * is a directory, all files in that directory are processed.
 * </p>
 * 
 * @param <K>
 *            type of key
 * @param <V>
 *            type of value
 */
public class SequenceFileProcessor<K extends WritableComparable, V extends Writable> {

  private Path mPath;
  private JobConf conf;
  private KeyValueProcess<K, V> mProcessor;
  private SequenceFile.Reader mReader;
  private K mKey;
  private V mValue;

  /**
   * Processes one or more <code>SequenceFile</code>s. The
   * {@link KeyValueProcess} is applied to every key-value pair in the file if
   * <code>path</code> denotes a file, or all files in the directory if
   * <code>path</code> denotes a directory.
   * 
   * @param <K1>
   *            type of key
   * @param <V1>
   *            type of value
   * @param path
   *            either a file or a directory
   * @param p
   *            the KeyValueProcess to apply
   * @return the KeyValueProcess applied
   */
  public static <K1 extends WritableComparable, V1 extends Writable> KeyValueProcess<K1, V1> process(
      String path, KeyValueProcess<K1, V1> p) {

    try {
      SequenceFileProcessor<K1, V1> processor = new SequenceFileProcessor<K1, V1>(
          path, p);
      processor.run();
    } catch (Exception e) {
      e.printStackTrace();
    }

    return p;
  }

  private SequenceFileProcessor(String location, KeyValueProcess<K, V> p)
      throws IOException {

    mPath = new Path(location);
    conf = new JobConf();

    mProcessor = p;

  }

  private void run() throws IOException {
    if (!FileSystem.get(conf).isFile(mPath)) {
      Path[] pa = new Path[] { mPath };
      Path p;
      // hadoop 0.17 -> listStatus();
      FileStatus[] fi = FileSystem.get(conf).listStatus(pa);
      for (int i =0 ; i<fi.length ; i++) {
        p = fi[i].getPath();
        // System.out.println("Applying to " + p);
        applyToFile(p);
      }
    } else {
      applyToFile(mPath);
    }

  }

  @SuppressWarnings("unchecked")
  private void applyToFile(Path path) throws IOException {
    mReader = new SequenceFile.Reader(FileSystem.get(conf), path, conf);

    try {
      mKey = (K) mReader.getKeyClass().newInstance();
      mValue = (V) mReader.getValueClass().newInstance();
    } catch (Exception e) {
      e.printStackTrace();
    }

    while (mReader.next(mKey, mValue) == true) {
      mProcessor.process(mKey, mValue);
    }

    mReader.close();
    mProcessor.report();
  }
}