testsequencefile.java

Hadoop是一个用于运行应用程序在大型集群的廉价硬件设备上的框架。Hadoop为应用程序透明的提供了一组稳定/可靠的接口和数据运动。在 Hadoop中实现了Google的MapReduce算法

/**
 * Copyright 2005 The Apache Software Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.io;

import java.io.*;
import java.util.*;
import junit.framework.TestCase;
import java.util.logging.*;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.*;


/** Support for flat files of binary key/value pairs. */
public class TestSequenceFile extends TestCase {
  private static Logger LOG = SequenceFile.LOG;

  private static Configuration conf = new Configuration();
  
  public TestSequenceFile(String name) { super(name); }

  /** Unit tests for SequenceFile. */
  public void testSequenceFile() throws Exception {
    int count = 1024 * 10;
    int megabytes = 1;
    int factor = 5;
    String file = System.getProperty("test.build.data",".") + "/test.seq";
 
    int seed = new Random().nextInt();

    FileSystem fs = new LocalFileSystem(new Configuration());
    try {
        //LOG.setLevel(Level.FINE);
        writeTest(fs, count, seed, file, false);
        readTest(fs, count, seed, file);

        sortTest(fs, count, megabytes, factor, false, file);
        checkSort(fs, count, seed, file);

        sortTest(fs, count, megabytes, factor, true, file);
        checkSort(fs, count, seed, file);

        mergeTest(fs, count, seed, file, false, factor, megabytes);
        checkSort(fs, count, seed, file);

        mergeTest(fs, count, seed, file, true, factor, megabytes);
        checkSort(fs, count, seed, file);
    } finally {
        fs.close();
    }
  }

  private static void writeTest(FileSystem fs, int count, int seed,
                                String file, boolean compress)
    throws IOException {
    new File(file).delete();
    LOG.fine("creating with " + count + " records");
    SequenceFile.Writer writer =
      new SequenceFile.Writer(fs, file, RandomDatum.class, RandomDatum.class,
                              compress);
    RandomDatum.Generator generator = new RandomDatum.Generator(seed);
    for (int i = 0; i < count; i++) {
      generator.next();
      RandomDatum key = generator.getKey();
      RandomDatum value = generator.getValue();

      writer.append(key, value);
    }
    writer.close();
  }

  private static void readTest(FileSystem fs, int count, int seed, String file)
    throws IOException {
    RandomDatum k = new RandomDatum();
    RandomDatum v = new RandomDatum();
    LOG.fine("reading " + count + " records");
    SequenceFile.Reader reader = new SequenceFile.Reader(fs, file, conf);
    RandomDatum.Generator generator = new RandomDatum.Generator(seed);
    for (int i = 0; i < count; i++) {
      generator.next();
      RandomDatum key = generator.getKey();
      RandomDatum value = generator.getValue();
      
      reader.next(k, v);
      
      if (!k.equals(key))
        throw new RuntimeException("wrong key at " + i);
      if (!v.equals(value))
        throw new RuntimeException("wrong value at " + i);
    }
    reader.close();
  }


  private static void sortTest(FileSystem fs, int count, int megabytes, 
                               int factor, boolean fast, String file)
    throws IOException {
    new File(file+".sorted").delete();
    SequenceFile.Sorter sorter = newSorter(fs, fast, megabytes, factor);
    LOG.fine("sorting " + count + " records");
    sorter.sort(file, file+".sorted");
    LOG.fine("done sorting " + count + " records");
  }

  private static void checkSort(FileSystem fs, int count, int seed, String file)
    throws IOException {
    LOG.fine("sorting " + count + " records in memory for check");
    RandomDatum.Generator generator = new RandomDatum.Generator(seed);
    SortedMap map = new TreeMap();
    for (int i = 0; i < count; i++) {
      generator.next();
      RandomDatum key = generator.getKey();
      RandomDatum value = generator.getValue();
      map.put(key, value);
    }

    LOG.fine("checking order of " + count + " records");
    RandomDatum k = new RandomDatum();
    RandomDatum v = new RandomDatum();
    Iterator iterator = map.entrySet().iterator();
    SequenceFile.Reader reader = new SequenceFile.Reader(fs, file + ".sorted", conf);
    for (int i = 0; i < count; i++) {
      Map.Entry entry = (Map.Entry)iterator.next();
      RandomDatum key = (RandomDatum)entry.getKey();
      RandomDatum value = (RandomDatum)entry.getValue();

      reader.next(k, v);

      if (!k.equals(key))
        throw new RuntimeException("wrong key at " + i);
      if (!v.equals(value))
        throw new RuntimeException("wrong value at " + i);
    }

    reader.close();
    LOG.fine("sucessfully checked " + count + " records");
  }

  private static void mergeTest(FileSystem fs, int count, int seed, 
                                String file, boolean fast, int factor, 
                                int megabytes)
    throws IOException {

    LOG.fine("creating "+factor+" files with "+count/factor+" records");

    SequenceFile.Writer[] writers = new SequenceFile.Writer[factor];
    String[] names = new String[factor];
    String[] sortedNames = new String[factor];
    
    for (int i = 0; i < factor; i++) {
      names[i] = file+"."+i;
      sortedNames[i] = names[i] + ".sorted";
      fs.delete(new File(names[i]));
      fs.delete(new File(sortedNames[i]));
      writers[i] =
        new SequenceFile.Writer(fs, names[i], RandomDatum.class,RandomDatum.class);
    }

    RandomDatum.Generator generator = new RandomDatum.Generator(seed);

    for (int i = 0; i < count; i++) {
      generator.next();
      RandomDatum key = generator.getKey();
      RandomDatum value = generator.getValue();

      writers[i%factor].append(key, value);
    }

    for (int i = 0; i < factor; i++)
      writers[i].close();

    for (int i = 0; i < factor; i++) {
      LOG.fine("sorting file " + i + " with " + count/factor + " records");
      newSorter(fs, fast, megabytes, factor).sort(names[i], sortedNames[i]);
    }

    LOG.fine("merging " + factor + " files with " + count/factor + " records");
    fs.delete(new File(file+".sorted"));
    newSorter(fs, fast, megabytes, factor).merge(sortedNames, file+".sorted");
  }

  private static SequenceFile.Sorter newSorter(FileSystem fs, 
                                               boolean fast,
                                               int megabytes, int factor) {
    SequenceFile.Sorter sorter = 
      fast
      ? new SequenceFile.Sorter(fs, new RandomDatum.Comparator(),RandomDatum.class, conf)
      : new SequenceFile.Sorter(fs, RandomDatum.class, RandomDatum.class, conf);
    sorter.setMemory(megabytes * 1024*1024);
    sorter.setFactor(factor);
    return sorter;
  }


  /** For debugging and testing. */
  public static void main(String[] args) throws Exception {
    int count = 1024 * 1024;
    int megabytes = 1;
    int factor = 10;
    boolean create = true;
    boolean check = false;
    boolean fast = false;
    boolean merge = false;
    boolean compress = false;
    String file = null;
    String usage = "Usage: SequenceFile (-local | -dfs <namenode:port>) [-count N] [-megabytes M] [-factor F] [-nocreate] [-check] [-fast] [-merge] [-compress] file";
    
    if (args.length == 0) {
        System.err.println(usage);
        System.exit(-1);
    }
    int i = 0;
    FileSystem fs = FileSystem.parseArgs(args, i, conf);      
    try {
      for (; i < args.length; i++) {       // parse command line
          if (args[i] == null) {
              continue;
          } else if (args[i].equals("-count")) {
              count = Integer.parseInt(args[++i]);
          } else if (args[i].equals("-megabytes")) {
              megabytes = Integer.parseInt(args[++i]);
          } else if (args[i].equals("-factor")) {
              factor = Integer.parseInt(args[++i]);
          } else if (args[i].equals("-nocreate")) {
              create = false;
          } else if (args[i].equals("-check")) {
              check = true;
          } else if (args[i].equals("-fast")) {
              fast = true;
          } else if (args[i].equals("-merge")) {
              merge = true;
          } else if (args[i].equals("-compress")) {
              compress = true;
          } else {
              // file is required parameter
              file = args[i];
          }
        }
        LOG.info("count = " + count);
        LOG.info("megabytes = " + megabytes);
        LOG.info("factor = " + factor);
        LOG.info("create = " + create);
        LOG.info("check = " + check);
        LOG.info("fast = " + fast);
        LOG.info("merge = " + merge);
        LOG.info("compress = " + compress);
        LOG.info("file = " + file);

        int seed = 0;
 
        LOG.setLevel(Level.FINE);

        if (create && !merge) {
            writeTest(fs, count, seed, file, compress);
            readTest(fs, count, seed, file);
        }

        if (merge) {
            mergeTest(fs, count, seed, file, fast, factor, megabytes);
        } else {
            sortTest(fs, count, megabytes, factor, fast, file);
        }
    
        if (check) {
            checkSort(fs, count, seed, file);
        }
      } finally {
          fs.close();
      }
  }
}