crunch-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jwi...@apache.org
Subject git commit: CRUNCH-209: Fix InputSplit bug that occurs with very large input directories
Date Thu, 23 May 2013 18:41:19 GMT
Updated Branches:
  refs/heads/master b24dc5804 -> 125f1b4e1


CRUNCH-209: Fix InputSplit bug that occurs with very large input directories


Project: http://git-wip-us.apache.org/repos/asf/crunch/repo
Commit: http://git-wip-us.apache.org/repos/asf/crunch/commit/125f1b4e
Tree: http://git-wip-us.apache.org/repos/asf/crunch/tree/125f1b4e
Diff: http://git-wip-us.apache.org/repos/asf/crunch/diff/125f1b4e

Branch: refs/heads/master
Commit: 125f1b4e143fbe16b18965e73a7a618effb68215
Parents: b24dc58
Author: Josh Wills <jwills@apache.org>
Authored: Wed May 22 17:26:24 2013 -0700
Committer: Josh Wills <jwills@apache.org>
Committed: Wed May 22 17:26:24 2013 -0700

----------------------------------------------------------------------
 .../crunch/impl/mr/run/CrunchInputFormat.java      |    4 +-
 .../crunch/impl/mr/run/CrunchInputSplit.java       |   49 ++++++++-----
 .../crunch/impl/mr/run/CrunchRecordReader.java     |   16 ++++-
 .../java/org/apache/crunch/io/FormatBundle.java    |   55 ++++++++++++++-
 4 files changed, 97 insertions(+), 27 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/crunch/blob/125f1b4e/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputFormat.java
----------------------------------------------------------------------
diff --git a/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputFormat.java
b/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputFormat.java
index eb5dd8a..cf3df81 100644
--- a/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputFormat.java
+++ b/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputFormat.java
@@ -23,7 +23,6 @@ import java.util.Map;
 
 import org.apache.crunch.io.CrunchInputs;
 import org.apache.crunch.io.FormatBundle;
-import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapreduce.InputFormat;
@@ -62,8 +61,7 @@ public class CrunchInputFormat<K, V> extends InputFormat<K, V>
{
         // and Mapper types by wrapping in a TaggedInputSplit.
         List<InputSplit> pathSplits = format.getSplits(jobCopy);
         for (InputSplit pathSplit : pathSplits) {
-          splits.add(new CrunchInputSplit(pathSplit, inputBundle.getFormatClass(),
-              nodeIndex, jobCopy.getConfiguration()));
+          splits.add(new CrunchInputSplit(pathSplit, inputBundle, nodeIndex, conf));
         }
       }
     }

http://git-wip-us.apache.org/repos/asf/crunch/blob/125f1b4e/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputSplit.java
----------------------------------------------------------------------
diff --git a/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputSplit.java
b/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputSplit.java
index b41062b..479a7c2 100644
--- a/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputSplit.java
+++ b/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchInputSplit.java
@@ -23,6 +23,8 @@ import java.io.DataOutput;
 import java.io.DataOutputStream;
 import java.io.IOException;
 
+import org.apache.crunch.io.FormatBundle;
+import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
@@ -33,11 +35,11 @@ import org.apache.hadoop.mapreduce.InputFormat;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.util.ReflectionUtils;
 
-class CrunchInputSplit extends InputSplit implements Writable {
+class CrunchInputSplit extends InputSplit implements Writable, Configurable {
 
   private InputSplit inputSplit;
-  private Class<? extends InputFormat<?, ?>> inputFormatClass;
   private int nodeIndex;
+  private FormatBundle<? extends InputFormat<?, ?>> bundle;
   private Configuration conf;
 
   public CrunchInputSplit() {
@@ -46,15 +48,24 @@ class CrunchInputSplit extends InputSplit implements Writable {
 
   public CrunchInputSplit(
       InputSplit inputSplit,
-      Class<? extends InputFormat<?, ?>> inputFormatClass,
+      FormatBundle<? extends InputFormat<?, ?>> bundle,
       int nodeIndex,
       Configuration conf) {
     this.inputSplit = inputSplit;
-    this.inputFormatClass = inputFormatClass;
+    this.bundle = bundle;
     this.nodeIndex = nodeIndex;
     this.conf = conf;
   }
 
+  @Override
+  public void setConf(Configuration conf) {
+    this.conf = conf;
+    if (bundle != null && conf != null) {
+      this.bundle.configure(conf);
+    }
+  }
+  
+  @Override
   public Configuration getConf() {
     return conf;
   }
@@ -68,7 +79,7 @@ class CrunchInputSplit extends InputSplit implements Writable {
   }
 
   public Class<? extends InputFormat<?, ?>> getInputFormatClass() {
-    return inputFormatClass;
+    return bundle.getFormatClass();
   }
 
   @Override
@@ -83,9 +94,10 @@ class CrunchInputSplit extends InputSplit implements Writable {
 
   public void readFields(DataInput in) throws IOException {
     nodeIndex = in.readInt();
-    conf = new Configuration();
-    conf.readFields(in);
-    inputFormatClass = (Class<? extends InputFormat<?, ?>>) readClass(in);
+    bundle = new FormatBundle();
+    bundle.setConf(conf);
+    bundle.readFields(in);
+    bundle.configure(conf); // yay bootstrap!
     Class<? extends InputSplit> inputSplitClass = (Class<? extends InputSplit>)
readClass(in);
     inputSplit = (InputSplit) ReflectionUtils.newInstance(inputSplitClass, conf);
     SerializationFactory factory = new SerializationFactory(conf);
@@ -94,23 +106,22 @@ class CrunchInputSplit extends InputSplit implements Writable {
     inputSplit = (InputSplit) deserializer.deserialize(inputSplit);
   }
 
-  private Class<?> readClass(DataInput in) throws IOException {
-    String className = Text.readString(in);
-    try {
-      return conf.getClassByName(className);
-    } catch (ClassNotFoundException e) {
-      throw new RuntimeException("readObject can't find class", e);
-    }
-  }
-
   public void write(DataOutput out) throws IOException {
     out.writeInt(nodeIndex);
-    conf.write(out);
-    Text.writeString(out, inputFormatClass.getName());
+    bundle.write(out);
     Text.writeString(out, inputSplit.getClass().getName());
     SerializationFactory factory = new SerializationFactory(conf);
     Serializer serializer = factory.getSerializer(inputSplit.getClass());
     serializer.open((DataOutputStream) out);
     serializer.serialize(inputSplit);
   }
+
+  private Class readClass(DataInput in) throws IOException {
+    String className = Text.readString(in);
+    try {
+      return conf.getClassByName(className);
+    } catch (ClassNotFoundException e) {
+      throw new RuntimeException("readObject can't find class", e);
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/crunch/blob/125f1b4e/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchRecordReader.java
----------------------------------------------------------------------
diff --git a/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchRecordReader.java
b/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchRecordReader.java
index fc8fb32..e5cbd95 100644
--- a/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchRecordReader.java
+++ b/crunch-core/src/main/java/org/apache/crunch/impl/mr/run/CrunchRecordReader.java
@@ -20,6 +20,7 @@ package org.apache.crunch.impl.mr.run;
 import java.io.IOException;
 
 import org.apache.crunch.hadoop.mapreduce.TaskAttemptContextFactory;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.InputFormat;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.RecordReader;
@@ -33,10 +34,15 @@ class CrunchRecordReader<K, V> extends RecordReader<K, V>
{
   public CrunchRecordReader(InputSplit inputSplit, final TaskAttemptContext context) throws
IOException,
       InterruptedException {
     CrunchInputSplit crunchSplit = (CrunchInputSplit) inputSplit;
+    Configuration conf = crunchSplit.getConf();
+    if (conf == null) {
+      conf = context.getConfiguration();
+      crunchSplit.setConf(conf);
+    }
     InputFormat<K, V> inputFormat = (InputFormat<K, V>) ReflectionUtils.newInstance(crunchSplit.getInputFormatClass(),
-        crunchSplit.getConf());
+        conf);
     this.delegate = inputFormat.createRecordReader(crunchSplit.getInputSplit(),
-        TaskAttemptContextFactory.create(crunchSplit.getConf(), context.getTaskAttemptID()));
+        TaskAttemptContextFactory.create(conf, context.getTaskAttemptID()));
   }
 
   @Override
@@ -62,9 +68,13 @@ class CrunchRecordReader<K, V> extends RecordReader<K, V> {
   @Override
   public void initialize(InputSplit inputSplit, TaskAttemptContext context) throws IOException,
InterruptedException {
     CrunchInputSplit crunchSplit = (CrunchInputSplit) inputSplit;
+    Configuration conf = crunchSplit.getConf();
+    if (conf == null) {
+      conf = context.getConfiguration();
+    }
     InputSplit delegateSplit = crunchSplit.getInputSplit();
     delegate.initialize(delegateSplit,
-        TaskAttemptContextFactory.create(crunchSplit.getConf(), context.getTaskAttemptID()));
+        TaskAttemptContextFactory.create(conf, context.getTaskAttemptID()));
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/crunch/blob/125f1b4e/crunch-core/src/main/java/org/apache/crunch/io/FormatBundle.java
----------------------------------------------------------------------
diff --git a/crunch-core/src/main/java/org/apache/crunch/io/FormatBundle.java b/crunch-core/src/main/java/org/apache/crunch/io/FormatBundle.java
index d969009..4796006 100644
--- a/crunch-core/src/main/java/org/apache/crunch/io/FormatBundle.java
+++ b/crunch-core/src/main/java/org/apache/crunch/io/FormatBundle.java
@@ -19,6 +19,8 @@ package org.apache.crunch.io;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.io.DataInput;
+import java.io.DataOutput;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
@@ -27,7 +29,10 @@ import java.util.Map;
 
 import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.lang.builder.HashCodeBuilder;
+import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.mapreduce.InputFormat;
 import org.apache.hadoop.mapreduce.OutputFormat;
 
@@ -41,11 +46,12 @@ import com.google.common.collect.Maps;
  * if they are the only format that exists in a particular MapReduce job, even
  * when we have multiple types of inputs and outputs within a single job.
  */
-public class FormatBundle<K> implements Serializable {
+public class FormatBundle<K> implements Serializable, Writable, Configurable {
 
   private Class<K> formatClass;
   private Map<String, String> extraConf;
-
+  private Configuration conf;
+  
   public static <T> FormatBundle<T> fromSerialized(String serialized, Class<T>
clazz) {
     ByteArrayInputStream bais = new ByteArrayInputStream(Base64.decodeBase64(serialized));
     try {
@@ -68,6 +74,10 @@ public class FormatBundle<K> implements Serializable {
     return new FormatBundle<T>(inputFormatClass);
   }
   
+  public FormatBundle() {
+    // For Writable support
+  }
+  
   private FormatBundle(Class<K> formatClass) {
     this.formatClass = formatClass;
     this.extraConf = Maps.newHashMap();
@@ -118,4 +128,45 @@ public class FormatBundle<K> implements Serializable {
     FormatBundle<K> oib = (FormatBundle<K>) other;
     return formatClass.equals(oib.formatClass) && extraConf.equals(oib.extraConf);
   }
+
+  @Override
+  public void readFields(DataInput in) throws IOException {
+    this.formatClass = readClass(in);
+    int ecSize = in.readInt();
+    this.extraConf = Maps.newHashMap();
+    for (int i = 0; i  < ecSize; i++) {
+      String key = Text.readString(in);
+      String value = Text.readString(in);
+      extraConf.put(key, value);
+    }
+  }
+
+  @Override
+  public void write(DataOutput out) throws IOException {
+    Text.writeString(out, formatClass.getName());
+    out.writeInt(extraConf.size());
+    for (Map.Entry<String, String> e : extraConf.entrySet()) {
+      Text.writeString(out, e.getKey());
+      Text.writeString(out, e.getValue());
+    }
+  }
+  
+  private Class readClass(DataInput in) throws IOException {
+    String className = Text.readString(in);
+    try {
+      return conf.getClassByName(className);
+    } catch (ClassNotFoundException e) {
+      throw new RuntimeException("readObject can't find class", e);
+    }
+  }
+
+  @Override
+  public Configuration getConf() {
+    return conf;
+  }
+
+  @Override
+  public void setConf(Configuration conf) {
+    this.conf = conf;
+  }
 }


Mime
View raw message