hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ha...@apache.org
Subject hadoop git commit: HADOOP-1381. The distance between sync blocks in SequenceFiles should be configurable rather than hard coded to 2000 bytes. Contributed by Harsh J.
Date Fri, 25 Nov 2016 16:52:41 GMT
Repository: hadoop
Updated Branches:
  refs/heads/trunk ee3d437a3 -> 07825f2b4


HADOOP-1381. The distance between sync blocks in SequenceFiles should be configurable rather
than hard coded to 2000 bytes. Contributed by Harsh J.


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/07825f2b
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/07825f2b
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/07825f2b

Branch: refs/heads/trunk
Commit: 07825f2b49384dbec92bfae87ea661cef9ffab49
Parents: ee3d437
Author: Harsh J <harsh@cloudera.com>
Authored: Wed Oct 26 20:04:33 2016 +0530
Committer: Harsh J <harsh@cloudera.com>
Committed: Fri Nov 25 22:22:23 2016 +0530

----------------------------------------------------------------------
 .../java/org/apache/hadoop/io/SequenceFile.java |  74 ++++++++----
 .../apache/hadoop/io/TestSequenceFileSync.java  | 113 +++++++++++++++----
 2 files changed, 146 insertions(+), 41 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/07825f2b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/SequenceFile.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/SequenceFile.java
b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/SequenceFile.java
index 2ac1389..c510ff7 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/SequenceFile.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/SequenceFile.java
@@ -24,6 +24,7 @@ import java.util.*;
 import java.rmi.server.UID;
 import java.security.MessageDigest;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.commons.logging.*;
 import org.apache.hadoop.util.Options;
 import org.apache.hadoop.fs.*;
@@ -146,7 +147,7 @@ import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_SKIP_CHECKSU
  *   </ul>
  * </li>
  * <li>
- * A sync-marker every few <code>100</code> bytes or so.
+ * A sync-marker every few <code>100</code> kilobytes or so.
  * </li>
  * </ul>
  *
@@ -165,7 +166,7 @@ import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_SKIP_CHECKSU
  *   </ul>
  * </li>
  * <li>
- * A sync-marker every few <code>100</code> bytes or so.
+ * A sync-marker every few <code>100</code> kilobytes or so.
  * </li>
  * </ul>
  * 
@@ -217,8 +218,11 @@ public class SequenceFile {
   private static final int SYNC_HASH_SIZE = 16;   // number of bytes in hash 
   private static final int SYNC_SIZE = 4+SYNC_HASH_SIZE; // escape + hash
 
-  /** The number of bytes between sync points.*/
-  public static final int SYNC_INTERVAL = 100*SYNC_SIZE; 
+  /**
+   * The number of bytes between sync points. 100 KB, default.
+   * Computed as 5 KB * 20 = 100 KB
+   */
+  public static final int SYNC_INTERVAL = 5 * 1024 * SYNC_SIZE; // 5KB*(16+4)
 
   /** 
    * The compression type used to compress key/value pairs in the 
@@ -856,6 +860,9 @@ public class SequenceFile {
     // starts and ends by scanning for this value.
     long lastSyncPos;                     // position of last sync
     byte[] sync;                          // 16 random bytes
+    @VisibleForTesting
+    int syncInterval;
+
     {
       try {                                       
         MessageDigest digester = MessageDigest.getInstance("MD5");
@@ -987,7 +994,16 @@ public class SequenceFile {
     private static Option filesystem(FileSystem fs) {
       return new SequenceFile.Writer.FileSystemOption(fs);
     }
-    
+
+    private static class SyncIntervalOption extends Options.IntegerOption
+        implements Option {
+      SyncIntervalOption(int val) {
+        // If a negative sync interval is provided,
+        // fall back to the default sync interval.
+        super(val < 0 ? SYNC_INTERVAL : val);
+      }
+    }
+
     public static Option bufferSize(int value) {
       return new BufferSizeOption(value);
     }
@@ -1032,11 +1048,15 @@ public class SequenceFile {
         CompressionCodec codec) {
       return new CompressionOption(value, codec);
     }
-    
+
+    public static Option syncInterval(int value) {
+      return new SyncIntervalOption(value);
+    }
+
     /**
      * Construct a uncompressed writer from a set of options.
      * @param conf the configuration to use
-     * @param options the options used when creating the writer
+     * @param opts the options used when creating the writer
      * @throws IOException if it fails
      */
     Writer(Configuration conf, 
@@ -1062,6 +1082,8 @@ public class SequenceFile {
         Options.getOption(MetadataOption.class, opts);
       CompressionOption compressionTypeOption =
         Options.getOption(CompressionOption.class, opts);
+      SyncIntervalOption syncIntervalOption =
+          Options.getOption(SyncIntervalOption.class, opts);
       // check consistency of options
       if ((fileOption == null) == (streamOption == null)) {
         throw new IllegalArgumentException("file or stream must be specified");
@@ -1163,7 +1185,12 @@ public class SequenceFile {
                                            "GzipCodec without native-hadoop " +
                                            "code!");
       }
-      init(conf, out, ownStream, keyClass, valueClass, codec, metadata);
+      this.syncInterval = (syncIntervalOption == null) ?
+          SYNC_INTERVAL :
+          syncIntervalOption.getValue();
+      init(
+          conf, out, ownStream, keyClass, valueClass,
+          codec, metadata, syncInterval);
     }
 
     /** Create the named file.
@@ -1176,7 +1203,7 @@ public class SequenceFile {
                   Class keyClass, Class valClass) throws IOException {
       this.compress = CompressionType.NONE;
       init(conf, fs.create(name), true, keyClass, valClass, null, 
-           new Metadata());
+           new Metadata(), SYNC_INTERVAL);
     }
     
     /** Create the named file with write-progress reporter.
@@ -1190,7 +1217,7 @@ public class SequenceFile {
                   Progressable progress, Metadata metadata) throws IOException {
       this.compress = CompressionType.NONE;
       init(conf, fs.create(name, progress), true, keyClass, valClass,
-           null, metadata);
+           null, metadata, SYNC_INTERVAL);
     }
     
     /** Create the named file with write-progress reporter. 
@@ -1206,7 +1233,7 @@ public class SequenceFile {
       this.compress = CompressionType.NONE;
       init(conf,
            fs.create(name, true, bufferSize, replication, blockSize, progress),
-           true, keyClass, valClass, null, metadata);
+           true, keyClass, valClass, null, metadata, SYNC_INTERVAL);
     }
 
     boolean isCompressed() { return compress != CompressionType.NONE; }
@@ -1234,18 +1261,21 @@ public class SequenceFile {
 
     /** Initialize. */
     @SuppressWarnings("unchecked")
-    void init(Configuration conf, FSDataOutputStream out, boolean ownStream,
-              Class keyClass, Class valClass,
-              CompressionCodec codec, Metadata metadata) 
+    void init(Configuration config, FSDataOutputStream outStream,
+              boolean ownStream, Class key, Class val,
+              CompressionCodec compCodec, Metadata meta,
+              int syncIntervalVal)
       throws IOException {
-      this.conf = conf;
-      this.out = out;
+      this.conf = config;
+      this.out = outStream;
       this.ownOutputStream = ownStream;
-      this.keyClass = keyClass;
-      this.valClass = valClass;
-      this.codec = codec;
-      this.metadata = metadata;
-      SerializationFactory serializationFactory = new SerializationFactory(conf);
+      this.keyClass = key;
+      this.valClass = val;
+      this.codec = compCodec;
+      this.metadata = meta;
+      this.syncInterval = syncIntervalVal;
+      SerializationFactory serializationFactory =
+          new SerializationFactory(config);
       this.keySerializer = serializationFactory.getSerializer(keyClass);
       if (this.keySerializer == null) {
         throw new IOException(
@@ -1366,7 +1396,7 @@ public class SequenceFile {
 
     synchronized void checkAndWriteSync() throws IOException {
       if (sync != null &&
-          out.getPos() >= lastSyncPos+SYNC_INTERVAL) { // time to emit sync
+          out.getPos() >= lastSyncPos+this.syncInterval) { // time to emit sync
         sync();
       }
     }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/07825f2b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileSync.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileSync.java
b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileSync.java
index bceb8af..363177b 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileSync.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestSequenceFileSync.java
@@ -27,13 +27,15 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.junit.Test;
 
+/** Tests sync based seek reads/write intervals inside SequenceFiles. */
 public class TestSequenceFileSync {
   private static final int NUMRECORDS = 2000;
   private static final int RECORDSIZE = 80;
-  private static final Random rand = new Random();
+  private static final Random RAND = new Random();
 
   private final static String REC_FMT = "%d RECORDID %d : ";
 
@@ -46,37 +48,110 @@ public class TestSequenceFileSync {
     reader.next(key, val);
     assertEquals(key.get(), expectedRecord);
     final String test = String.format(REC_FMT, expectedRecord, expectedRecord);
-    assertEquals("Invalid value " + val, 0, val.find(test, 0));
+    assertEquals(
+        "Invalid value in iter " + iter + ": " + val,
+        0,
+        val.find(test, 0));
+  }
+
+  @Test
+  public void testDefaultSyncInterval() throws IOException {
+    // Uses the default sync interval of 100 KB
+    final Configuration conf = new Configuration();
+    final FileSystem fs = FileSystem.getLocal(conf);
+    final Path path = new Path(GenericTestUtils.getTempPath(
+            "sequencefile.sync.test"));
+    final IntWritable input = new IntWritable();
+    final Text val = new Text();
+    SequenceFile.Writer writer = new SequenceFile.Writer(
+        conf,
+        SequenceFile.Writer.file(path),
+        SequenceFile.Writer.compression(CompressionType.NONE),
+        SequenceFile.Writer.keyClass(IntWritable.class),
+        SequenceFile.Writer.valueClass(Text.class)
+    );
+    try {
+      writeSequenceFile(writer, NUMRECORDS*4);
+      for (int i = 0; i < 5; i++) {
+        final SequenceFile.Reader reader;
+
+        //try different SequenceFile.Reader constructors
+        if (i % 2 == 0) {
+          final int buffersize = conf.getInt("io.file.buffer.size", 4096);
+          reader = new SequenceFile.Reader(conf,
+              SequenceFile.Reader.file(path),
+              SequenceFile.Reader.bufferSize(buffersize));
+        } else {
+          final FSDataInputStream in = fs.open(path);
+          final long length = fs.getFileStatus(path).getLen();
+          reader = new SequenceFile.Reader(conf,
+              SequenceFile.Reader.stream(in),
+              SequenceFile.Reader.start(0L),
+              SequenceFile.Reader.length(length));
+        }
+
+        try {
+          forOffset(reader, input, val, i, 0, 0);
+          forOffset(reader, input, val, i, 65, 0);
+          // There would be over 1000 records within
+          // this sync interval
+          forOffset(reader, input, val, i, 2000, 1101);
+          forOffset(reader, input, val, i, 0, 0);
+        } finally {
+          reader.close();
+        }
+      }
+    } finally {
+      fs.delete(path, false);
+    }
   }
 
   @Test
   public void testLowSyncpoint() throws IOException {
+    // Uses a smaller sync interval of 2000 bytes
     final Configuration conf = new Configuration();
     final FileSystem fs = FileSystem.getLocal(conf);
     final Path path = new Path(GenericTestUtils.getTempPath(
         "sequencefile.sync.test"));
     final IntWritable input = new IntWritable();
     final Text val = new Text();
-    SequenceFile.Writer writer = new SequenceFile.Writer(fs, conf, path,
-        IntWritable.class, Text.class);
+    SequenceFile.Writer writer = new SequenceFile.Writer(
+        conf,
+        SequenceFile.Writer.file(path),
+        SequenceFile.Writer.compression(CompressionType.NONE),
+        SequenceFile.Writer.keyClass(IntWritable.class),
+        SequenceFile.Writer.valueClass(Text.class),
+        SequenceFile.Writer.syncInterval(20*100)
+    );
+    // Ensure the custom sync interval value is set
+    assertEquals(writer.syncInterval, 20*100);
     try {
       writeSequenceFile(writer, NUMRECORDS);
-      for (int i = 0; i < 5 ; i++) {
-       final SequenceFile.Reader reader;
+      for (int i = 0; i < 5; i++) {
+        final SequenceFile.Reader reader;
        
-       //try different SequenceFile.Reader constructors
-       if (i % 2 == 0) {
-         reader = new SequenceFile.Reader(fs, path, conf);
-       } else {
-         final FSDataInputStream in = fs.open(path);
-         final long length = fs.getFileStatus(path).getLen();
-         final int buffersize = conf.getInt("io.file.buffer.size", 4096);
-         reader = new SequenceFile.Reader(in, buffersize, 0L, length, conf);
-       }
+        //try different SequenceFile.Reader constructors
+        if (i % 2 == 0) {
+          final int bufferSize = conf.getInt("io.file.buffer.size", 4096);
+          reader = new SequenceFile.Reader(
+              conf,
+              SequenceFile.Reader.file(path),
+              SequenceFile.Reader.bufferSize(bufferSize));
+        } else {
+          final FSDataInputStream in = fs.open(path);
+          final long length = fs.getFileStatus(path).getLen();
+          reader = new SequenceFile.Reader(
+              conf,
+              SequenceFile.Reader.stream(in),
+              SequenceFile.Reader.start(0L),
+              SequenceFile.Reader.length(length));
+        }
 
-       try {
+        try {
           forOffset(reader, input, val, i, 0, 0);
           forOffset(reader, input, val, i, 65, 0);
+          // There would be only a few records within
+          // this sync interval
           forOffset(reader, input, val, i, 2000, 21);
           forOffset(reader, input, val, i, 0, 0);
         } finally {
@@ -88,7 +163,7 @@ public class TestSequenceFileSync {
     }
   }
 
-  public static void writeSequenceFile(SequenceFile.Writer writer,
+  private static void writeSequenceFile(SequenceFile.Writer writer,
       int numRecords) throws IOException {
     final IntWritable key = new IntWritable();
     final Text val = new Text();
@@ -100,13 +175,13 @@ public class TestSequenceFileSync {
     writer.close();
   }
 
-  static void randomText(Text val, int id, int recordSize) {
+  private static void randomText(Text val, int id, int recordSize) {
     val.clear();
     final StringBuilder ret = new StringBuilder(recordSize);
     ret.append(String.format(REC_FMT, id, id));
     recordSize -= ret.length();
     for (int i = 0; i < recordSize; ++i) {
-      ret.append(rand.nextInt(9));
+      ret.append(RAND.nextInt(9));
     }
     val.set(ret.toString());
   }


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message