svn commit: r1056517 - in /avro/trunk: ./ lang/java/avro/src/main/java/org/apache/avro/file/ lang/java/mapred/src/main/java/org/apache/avro/mapred/ lang/java/mapred/src/test/java/org/apache/avro/mapred/

[cu...@apache.org]

Author: cutting
Date: Fri Jan  7 21:19:36 2011
New Revision: 1056517

URL: http://svn.apache.org/viewvc?rev=1056517&view=rev
Log:
AVRO-719. Java: Permit MapReduce programs to alter output file sync interval.  Contributed by Joe Crobak.

Added:
    avro/trunk/lang/java/mapred/src/test/java/org/apache/avro/mapred/TestAvroOutputFormat.java
Modified:
    avro/trunk/CHANGES.txt
    avro/trunk/lang/java/avro/src/main/java/org/apache/avro/file/DataFileConstants.java
    avro/trunk/lang/java/mapred/src/main/java/org/apache/avro/mapred/AvroOutputFormat.java

Modified: avro/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/avro/trunk/CHANGES.txt?rev=1056517&r1=1056516&r2=1056517&view=diff
==============================================================================
--- avro/trunk/CHANGES.txt (original)
+++ avro/trunk/CHANGES.txt Fri Jan  7 21:19:36 2011
@@ -85,6 +85,9 @@ Avro 1.5.0 (unreleased)
     AVRO-723. Java: Pass error messages for unexpected RPC exceptions
     through to client. (Stephen Gargan via cutting)
 
+    AVRO-719. Java: Permit MapReduce programs to alter output file
+    sync interval.  (Joe Crobak via cutting)
+
   BUG FIXES
 
     AVRO-675. C: Bytes and fixed setters don't update datum size.

Modified: avro/trunk/lang/java/avro/src/main/java/org/apache/avro/file/DataFileConstants.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/avro/src/main/java/org/apache/avro/file/DataFileConstants.java?rev=1056517&r1=1056516&r2=1056517&view=diff
==============================================================================
--- avro/trunk/lang/java/avro/src/main/java/org/apache/avro/file/DataFileConstants.java (original)
+++ avro/trunk/lang/java/avro/src/main/java/org/apache/avro/file/DataFileConstants.java Fri Jan  7 21:19:36 2011
@@ -21,7 +21,9 @@ package org.apache.avro.file;
 /**
  * Constants used in data files.
  */
-class DataFileConstants {
+public class DataFileConstants {
+  private DataFileConstants() {}                  // no public ctor
+
   public static final byte VERSION = 1;
   public static final byte[] MAGIC = new byte[] {
     (byte)'O', (byte)'b', (byte)'j', VERSION

Modified: avro/trunk/lang/java/mapred/src/main/java/org/apache/avro/mapred/AvroOutputFormat.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/mapred/src/main/java/org/apache/avro/mapred/AvroOutputFormat.java?rev=1056517&r1=1056516&r2=1056517&view=diff
==============================================================================
--- avro/trunk/lang/java/mapred/src/main/java/org/apache/avro/mapred/AvroOutputFormat.java (original)
+++ avro/trunk/lang/java/mapred/src/main/java/org/apache/avro/mapred/AvroOutputFormat.java Fri Jan  7 21:19:36 2011
@@ -35,6 +35,7 @@ import org.apache.avro.Schema;
 import org.apache.avro.reflect.ReflectDatumWriter;
 import org.apache.avro.file.DataFileWriter;
 import org.apache.avro.file.CodecFactory;
+import static org.apache.avro.file.DataFileConstants.DEFAULT_SYNC_INTERVAL;
 
 /** An {@link org.apache.hadoop.mapred.OutputFormat} for Avro data files. */
 public class AvroOutputFormat <T>
@@ -46,6 +47,9 @@ public class AvroOutputFormat <T>
   /** The configuration key for Avro deflate level. */
   public static final String DEFLATE_LEVEL_KEY = "avro.mapred.deflate.level";
 
+  /** The configuration key for Avro sync interval. */
+  public static final String SYNC_INTERVAL_KEY = "avro.mapred.sync.interval";
+
   /** The default deflate level. */
   public static final int DEFAULT_DEFLATE_LEVEL = 1;
 
@@ -55,6 +59,12 @@ public class AvroOutputFormat <T>
     job.setInt(DEFLATE_LEVEL_KEY, level);
   }
 
+  /** Set the sync interval to be used by the underlying {@link DataFileWriter}.*/
+  public static void setSyncInterval(JobConf job, int syncIntervalInBytes) {
+    job.setInt(SYNC_INTERVAL_KEY, syncIntervalInBytes);
+  }
+
+  @Override
   public RecordWriter<AvroWrapper<T>, NullWritable>
     getRecordWriter(FileSystem ignore, JobConf job,
                     String name, Progressable prog)
@@ -73,6 +83,8 @@ public class AvroOutputFormat <T>
       writer.setCodec(CodecFactory.deflateCodec(level));
     }
 
+    writer.setSyncInterval(job.getInt(SYNC_INTERVAL_KEY, DEFAULT_SYNC_INTERVAL));
+
     // copy metadata from job
     for (Map.Entry<String,String> e : job) {
       if (e.getKey().startsWith(AvroJob.TEXT_PREFIX))

Added: avro/trunk/lang/java/mapred/src/test/java/org/apache/avro/mapred/TestAvroOutputFormat.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/mapred/src/test/java/org/apache/avro/mapred/TestAvroOutputFormat.java?rev=1056517&view=auto
==============================================================================
--- avro/trunk/lang/java/mapred/src/test/java/org/apache/avro/mapred/TestAvroOutputFormat.java (added)
+++ avro/trunk/lang/java/mapred/src/test/java/org/apache/avro/mapred/TestAvroOutputFormat.java Fri Jan  7 21:19:36 2011
@@ -0,0 +1,36 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.avro.mapred;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.mapred.JobConf;
+import org.junit.Test;
+
+public class TestAvroOutputFormat extends TestCase {
+  @Test
+  public void testSetSyncInterval() {
+    JobConf jobConf = new JobConf();
+    int newSyncInterval = 100000;
+    AvroOutputFormat.setSyncInterval(jobConf, newSyncInterval);
+
+    assertEquals(newSyncInterval, jobConf.getInt(
+        AvroOutputFormat.SYNC_INTERVAL_KEY, -1));
+  }
+
+}