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 <[email protected]> Authored: Wed May 22 17:26:24 2013 -0700 Committer: Josh Wills <[email protected]> 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; + } }
