You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by bi...@apache.org on 2014/08/29 06:53:34 UTC
git commit: TEZ-1511. MROutputConfigBuilder sets OutputFormat as
String class if OutputFormat is not provided (bikas)
Repository: tez
Updated Branches:
refs/heads/master db4161b03 -> 7f70f9ddd
TEZ-1511. MROutputConfigBuilder sets OutputFormat as String class if OutputFormat is not provided (bikas)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/7f70f9dd
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/7f70f9dd
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/7f70f9dd
Branch: refs/heads/master
Commit: 7f70f9ddd433b4d26eb79934396ce5102adf759c
Parents: db4161b
Author: Bikas Saha <bi...@apache.org>
Authored: Thu Aug 28 21:53:22 2014 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Thu Aug 28 21:53:22 2014 -0700
----------------------------------------------------------------------
.../tez/mapreduce/hadoop/MRJobConfig.java | 4 +++
.../org/apache/tez/mapreduce/input/MRInput.java | 36 ++++++++++++--------
.../apache/tez/mapreduce/output/MROutput.java | 35 +++++++++++--------
.../tez/mapreduce/examples/UnionExample.java | 15 +++++---
4 files changed, 57 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/7f70f9dd/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRJobConfig.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRJobConfig.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRJobConfig.java
index 456e8f3..d44be30 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRJobConfig.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRJobConfig.java
@@ -31,6 +31,10 @@ public interface MRJobConfig {
// Put all of the attribute names in here so that Job and JobContext are
// consistent.
public static final String INPUT_FORMAT_CLASS_ATTR = "mapreduce.job.inputformat.class";
+
+ public static final String NEW_API_MAPPER_CONFIG = "mapred.mapper.new-api";
+
+ public static final String NEW_API_REDUCER_CONFIG = "mapred.reducer.new-api";
public static final String MAP_CLASS_ATTR = "mapreduce.job.map.class";
http://git-wip-us.apache.org/repos/asf/tez/blob/7f70f9dd/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java
index d541e7e..ddb001c 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java
@@ -26,6 +26,7 @@ import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import com.google.protobuf.ByteString;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -38,7 +39,6 @@ import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.split.JobSplit.TaskSplitIndex;
import org.apache.hadoop.mapreduce.split.JobSplit.TaskSplitMetaInfo;
import org.apache.hadoop.security.Credentials;
-import org.apache.hadoop.util.ReflectionUtils;
import org.apache.tez.client.TezClientUtils;
import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.dag.api.DataSourceDescriptor;
@@ -80,7 +80,7 @@ import com.google.common.collect.Lists;
*/
@Public
public class MRInput extends MRInputBase {
-
+
/**
* Helper class to configure {@link MRInput}
*
@@ -97,28 +97,36 @@ public class MRInput extends MRInputBase {
String inputPaths = null;
InputInitializerDescriptor customInitializerDescriptor = null;
- MRInputConfigBuilder(Configuration conf, Class<?> inputFormat) {
+ MRInputConfigBuilder(Configuration conf, Class<?> inputFormatParam) {
this.conf = conf;
- if (inputFormat != null) {
+ if (inputFormatParam != null) {
inputFormatProvided = true;
- this.inputFormat = inputFormat;
- if (org.apache.hadoop.mapred.InputFormat.class.isAssignableFrom(inputFormat)) {
+ this.inputFormat = inputFormatParam;
+ if (org.apache.hadoop.mapred.InputFormat.class.isAssignableFrom(inputFormatParam)) {
useNewApi = false;
- } else if (org.apache.hadoop.mapreduce.InputFormat.class.isAssignableFrom(inputFormat)) {
+ } else if (org.apache.hadoop.mapreduce.InputFormat.class.isAssignableFrom(inputFormatParam)) {
useNewApi = true;
} else {
throw new TezUncheckedException("inputFormat must be assignable from either " +
"org.apache.hadoop.mapred.InputFormat or " +
"org.apache.hadoop.mapreduce.InputFormat" +
- " Given: " + inputFormat.getName());
+ " Given: " + inputFormatParam.getName());
}
} else {
inputFormatProvided = false;
- useNewApi = conf.getBoolean("mapred.mapper.new-api", true);
- if (useNewApi) {
- this.inputFormat = ReflectionUtils.getClass(conf.get(MRJobConfig.INPUT_FORMAT_CLASS_ATTR));
- } else {
- this.inputFormat = ReflectionUtils.getClass(conf.get("mapred.input.format.class"));
+ useNewApi = conf.getBoolean(MRJobConfig.NEW_API_MAPPER_CONFIG, true);
+ try {
+ if (useNewApi) {
+ this.inputFormat = conf.getClassByName(conf.get(MRJobConfig.INPUT_FORMAT_CLASS_ATTR));
+ Preconditions.checkState(org.apache.hadoop.mapreduce.InputFormat.class
+ .isAssignableFrom(this.inputFormat));
+ } else {
+ this.inputFormat = conf.getClassByName(conf.get("mapred.input.format.class"));
+ Preconditions.checkState(org.apache.hadoop.mapred.InputFormat.class
+ .isAssignableFrom(this.inputFormat));
+ }
+ } catch (ClassNotFoundException e) {
+ throw new TezUncheckedException(e);
}
initializeInputPath();
}
@@ -292,7 +300,7 @@ public class MRInput extends MRInputBase {
private void setupBasicConf(Configuration inputConf) {
if (inputFormatProvided) {
- inputConf.setBoolean("mapred.mapper.new-api", useNewApi);
+ inputConf.setBoolean(MRJobConfig.NEW_API_MAPPER_CONFIG, useNewApi);
if (useNewApi) {
inputConf.set(MRJobConfig.INPUT_FORMAT_CLASS_ATTR, inputFormat.getName());
} else {
http://git-wip-us.apache.org/repos/asf/tez/blob/7f70f9dd/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
index fd0ea46..db9fbea 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
@@ -43,7 +43,6 @@ import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.util.ReflectionUtils;
import org.apache.tez.client.TezClientUtils;
import org.apache.tez.common.TezUtils;
import org.apache.tez.common.counters.TaskCounter;
@@ -92,28 +91,36 @@ public class MROutput extends AbstractLogicalOutput {
String outputPath;
boolean doCommit = true;
- private MROutputConfigBuilder(Configuration conf, Class<?> outputFormat) {
+ private MROutputConfigBuilder(Configuration conf, Class<?> outputFormatParam) {
this.conf = conf;
- if (outputFormat != null) {
+ if (outputFormatParam != null) {
outputFormatProvided = true;
- this.outputFormat = outputFormat;
- if (org.apache.hadoop.mapred.OutputFormat.class.isAssignableFrom(outputFormat)) {
+ this.outputFormat = outputFormatParam;
+ if (org.apache.hadoop.mapred.OutputFormat.class.isAssignableFrom(outputFormatParam)) {
useNewApi = false;
- } else if (org.apache.hadoop.mapreduce.OutputFormat.class.isAssignableFrom(outputFormat)) {
+ } else if (org.apache.hadoop.mapreduce.OutputFormat.class.isAssignableFrom(outputFormatParam)) {
useNewApi = true;
} else {
throw new TezUncheckedException("outputFormat must be assignable from either " +
"org.apache.hadoop.mapred.OutputFormat or " +
"org.apache.hadoop.mapreduce.OutputFormat" +
- " Given: " + outputFormat.getName());
+ " Given: " + outputFormatParam.getName());
}
} else {
outputFormatProvided = false;
- useNewApi = conf.getBoolean("mapred.mapper.new-api", true);
- if (useNewApi) {
- this.outputFormat = ReflectionUtils.getClass(conf.get(MRJobConfig.OUTPUT_FORMAT_CLASS_ATTR));
- } else {
- this.outputFormat = ReflectionUtils.getClass(conf.get("mapred.output.format.class"));
+ useNewApi = conf.getBoolean(MRJobConfig.NEW_API_REDUCER_CONFIG, true);
+ try {
+ if (useNewApi) {
+ this.outputFormat = conf.getClassByName(conf.get(MRJobConfig.OUTPUT_FORMAT_CLASS_ATTR));
+ Preconditions.checkState(org.apache.hadoop.mapreduce.OutputFormat.class
+ .isAssignableFrom(this.outputFormat));
+ } else {
+ this.outputFormat = conf.getClassByName(conf.get("mapred.output.format.class"));
+ Preconditions.checkState(org.apache.hadoop.mapred.OutputFormat.class
+ .isAssignableFrom(this.outputFormat));
+ }
+ } catch (ClassNotFoundException e) {
+ throw new TezUncheckedException(e);
}
initializeOutputPath();
}
@@ -210,7 +217,7 @@ public class MROutput extends AbstractLogicalOutput {
*/
private UserPayload createUserPayload() {
if (outputFormatProvided) {
- conf.setBoolean("mapred.reducer.new-api", useNewApi);
+ conf.setBoolean(MRJobConfig.NEW_API_REDUCER_CONFIG, useNewApi);
if (useNewApi) {
conf.set(MRJobConfig.OUTPUT_FORMAT_CLASS_ATTR, outputFormat.getName());
} else {
@@ -346,7 +353,7 @@ public class MROutput extends AbstractLogicalOutput {
newApiTaskAttemptContext = createTaskAttemptContext(taskAttemptId);
try {
newOutputFormat =
- ReflectionUtils.newInstance(
+ org.apache.hadoop.util.ReflectionUtils.newInstance(
newApiTaskAttemptContext.getOutputFormatClass(), jobConf);
} catch (ClassNotFoundException cnfe) {
throw new IOException(cnfe);
http://git-wip-us.apache.org/repos/asf/tez/blob/7f70f9dd/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
index 56fb1a3..7688335 100644
--- a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
+++ b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
@@ -32,6 +32,8 @@ import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileAlreadyExistsException;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.tez.client.TezClient;
@@ -168,9 +170,10 @@ public class UnionExample {
int numMaps = -1;
Configuration inputConf = new Configuration(tezConf);
- MRInput.MRInputConfigBuilder configurer = MRInput.createConfigBuilder(inputConf,
- TextInputFormat.class,
- inputPath);
+ inputConf.setBoolean("mapred.mapper.new-api", false);
+ inputConf.set("mapred.input.format.class", TextInputFormat.class.getName());
+ inputConf.set(FileInputFormat.INPUT_DIR, inputPath);
+ MRInput.MRInputConfigBuilder configurer = MRInput.createConfigBuilder(inputConf, null);
DataSourceDescriptor dataSource = configurer.generateSplitsInAM(false).build();
Vertex mapVertex1 = Vertex.create("map1", ProcessorDescriptor.create(
@@ -186,8 +189,10 @@ public class UnionExample {
UnionProcessor.class.getName()), 1);
Configuration outputConf = new Configuration(tezConf);
- DataSinkDescriptor od = MROutput.createConfigBuilder(outputConf,
- TextOutputFormat.class, outputPath).build();
+ outputConf.setBoolean("mapred.reducer.new-api", false);
+ outputConf.set("mapred.output.format.class", TextOutputFormat.class.getName());
+ outputConf.set(FileOutputFormat.OUTDIR, outputPath);
+ DataSinkDescriptor od = MROutput.createConfigBuilder(outputConf, null).build();
checkerVertex.addDataSink("union", od);