You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ro...@apache.org on 2014/07/16 02:01:46 UTC
svn commit: r1610885 - in /pig/trunk: CHANGES.txt
src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java
src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java
Author: rohini
Date: Wed Jul 16 00:01:45 2014
New Revision: 1610885
URL: http://svn.apache.org/r1610885
Log:
PIG-4058: Use single config in Tez for input and output (rohini)
Modified:
pig/trunk/CHANGES.txt
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1610885&r1=1610884&r2=1610885&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Wed Jul 16 00:01:45 2014
@@ -42,6 +42,8 @@ OPTIMIZATIONS
BUG FIXES
+PIG-4058: Use single config in Tez for input and output (rohini)
+
PIG-3886: UdfDistributedCache_1 fails in tez branch (cheolsoo)
PIG-4055 Build broke after TEZ-1130 API rename (knoguchi)
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java?rev=1610885&r1=1610884&r2=1610885&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java Wed Jul 16 00:01:45 2014
@@ -283,7 +283,6 @@ public class TezDagBuilder extends TezOp
for (POLocalRearrangeTez lr : lrs) {
if (lr.getOutputKey().equals(to.getOperatorKey().toString())) {
byte keyType = lr.getKeyType();
- setIntermediateInputKeyValue(keyType, conf, to, lr.isConnectedToPackage());
setIntermediateOutputKeyValue(keyType, conf, to, lr.isConnectedToPackage());
// In case of secondary key sort, main key type is the actual key type
conf.set("pig.reduce.key.type", Byte.toString(lr.getMainKeyType()));
@@ -295,23 +294,17 @@ public class TezDagBuilder extends TezOp
MRPartitioner.class.getName());
if (edge.getIntermediateOutputKeyClass() != null) {
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_KEY_CLASS,
- edge.getIntermediateOutputKeyClass());
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_CLASS,
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
edge.getIntermediateOutputKeyClass());
}
if (edge.getIntermediateOutputValueClass() != null) {
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_VALUE_CLASS,
- edge.getIntermediateOutputValueClass());
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_VALUE_CLASS,
+ conf.set(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS,
edge.getIntermediateOutputValueClass());
}
if (edge.getIntermediateOutputKeyComparatorClass() != null) {
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_KEY_COMPARATOR_CLASS,
- edge.getIntermediateOutputKeyComparatorClass());
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_COMPARATOR_CLASS,
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
edge.getIntermediateOutputKeyComparatorClass());
}
@@ -330,17 +323,14 @@ public class TezDagBuilder extends TezOp
ObjectSerializer.serialize(edge.getSecondarySortOrder()));
conf.set(org.apache.hadoop.mapreduce.MRJobConfig.PARTITIONER_CLASS_ATTR,
SecondaryKeyPartitioner.class.getName());
- // In MR - job.setSortComparatorClass() or MRJobConfig.KEY_COMPARATOR
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_KEY_COMPARATOR_CLASS,
- PigSecondaryKeyComparator.class.getName());
- // In MR - job.setOutputKeyClass() or MRJobConfig.OUTPUT_KEY_CLASS
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_KEY_CLASS, NullableTuple.class.getName());
// These needs to be on the vertex as well for POShuffleTezLoad to pick it up.
// Tez framework also expects this to be per vertex and not edge. IFile.java picks
// up keyClass and valueClass from vertex config. TODO - check with Tez folks
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_COMPARATOR_CLASS,
+ // In MR - job.setSortComparatorClass() or MRJobConfig.KEY_COMPARATOR
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
PigSecondaryKeyComparator.class.getName());
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_CLASS, NullableTuple.class.getName());
+ // In MR - job.setOutputKeyClass() or MRJobConfig.OUTPUT_KEY_CLASS
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, NullableTuple.class.getName());
setGroupingComparator(conf, PigSecondaryKeyGroupComparator.class.getName());
}
@@ -370,8 +360,6 @@ public class TezDagBuilder extends TezOp
private void addCombiner(PhysicalPlan combinePlan, TezOperator pkgTezOp,
Configuration conf) throws IOException {
POPackage combPack = (POPackage) combinePlan.getRoots().get(0);
- setIntermediateInputKeyValue(combPack.getPkgr().getKeyType(), conf, pkgTezOp);
-
POLocalRearrange combRearrange = (POLocalRearrange) combinePlan
.getLeaves().get(0);
setIntermediateOutputKeyValue(combRearrange.getKeyType(), conf, pkgTezOp);
@@ -470,7 +458,7 @@ public class TezDagBuilder extends TezOp
byte keyType = pack.getPkgr().getKeyType();
tezOp.plan.remove(pack);
payloadConf.set("pig.reduce.package", ObjectSerializer.serialize(pack));
- setIntermediateInputKeyValue(keyType, payloadConf, tezOp);
+ setIntermediateOutputKeyValue(keyType, payloadConf, tezOp);
POShuffleTezLoad newPack;
newPack = new POShuffleTezLoad(pack);
if (tezOp.isSkewedJoin()) {
@@ -510,7 +498,7 @@ public class TezDagBuilder extends TezOp
}
}
- setIntermediateInputKeyValue(pack.getPkgr().getKeyType(), payloadConf, tezOp);
+ setIntermediateOutputKeyValue(pack.getPkgr().getKeyType(), payloadConf, tezOp);
} else if (roots.size() == 1 && roots.get(0) instanceof POIdentityInOutTez) {
POIdentityInOutTez identityInOut = (POIdentityInOutTez) roots.get(0);
// TODO Need to fix multiple input key mapping
@@ -919,56 +907,34 @@ public class TezDagBuilder extends TezOp
return stores;
}
+ private void setIntermediateOutputKeyValue(byte keyType, Configuration conf, TezOperator tezOp)
+ throws JobCreationException, ExecException {
+ setIntermediateOutputKeyValue(keyType, conf, tezOp, true);
+ }
+
@SuppressWarnings("rawtypes")
- private void setIntermediateInputKeyValue(byte keyType, Configuration conf, TezOperator tezOp,
+ private void setIntermediateOutputKeyValue(byte keyType, Configuration conf, TezOperator tezOp,
boolean isConnectedToPackage) throws JobCreationException, ExecException {
if (tezOp != null && tezOp.isUseSecondaryKey() && isConnectedToPackage) {
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_CLASS,
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
NullableTuple.class.getName());
} else if (tezOp != null && tezOp.isSkewedJoin() && isConnectedToPackage) {
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_CLASS,
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
NullablePartitionWritable.class.getName());
} else {
Class<? extends WritableComparable> keyClass = HDataType
.getWritableComparableTypes(keyType).getClass();
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_CLASS,
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
keyClass.getName());
}
- selectInputComparator(conf, keyType, tezOp);
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_VALUE_CLASS,
- NullableTuple.class.getName());
- }
-
- private void setIntermediateInputKeyValue(byte keyType, Configuration conf, TezOperator tezOp)
- throws JobCreationException, ExecException {
- setIntermediateInputKeyValue(keyType, conf, tezOp, true);
- }
-
- @SuppressWarnings("rawtypes")
- private void setIntermediateOutputKeyValue(byte keyType, Configuration conf, TezOperator tezOp,
- boolean isConnectedToPackage) throws JobCreationException, ExecException {
- Class<? extends WritableComparable> keyClass = HDataType
- .getWritableComparableTypes(keyType).getClass();
- if (tezOp != null && tezOp.isSkewedJoin() && isConnectedToPackage) {
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_KEY_CLASS,
- NullablePartitionWritable.class.getName());
- } else {
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_KEY_CLASS,
- keyClass.getName());
- }
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_VALUE_CLASS,
+ conf.set(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS,
NullableTuple.class.getName());
conf.set(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS,
MRPartitioner.class.getName());
selectOutputComparator(keyType, conf, tezOp);
}
- private void setIntermediateOutputKeyValue(byte keyType, Configuration conf, TezOperator tezOp)
- throws JobCreationException, ExecException {
- setIntermediateOutputKeyValue(keyType, conf, tezOp, true);
- }
-
private static Class<? extends WritableComparator> comparatorForKeyType(byte keyType, boolean hasOrderBy)
throws JobCreationException {
@@ -1088,25 +1054,25 @@ public class TezDagBuilder extends TezOp
}
}
- void selectInputComparator(Configuration conf, byte keyType, TezOperator tezOp)
+ void selectOutputComparator(byte keyType, Configuration conf, TezOperator tezOp)
throws JobCreationException {
// TODO: Handle sorting like in JobControlCompiler
// TODO: Group comparators as in JobControlCompiler
if (tezOp != null && tezOp.isUseSecondaryKey()) {
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_COMPARATOR_CLASS,
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
PigSecondaryKeyComparator.class.getName());
setGroupingComparator(conf, PigSecondaryKeyGroupComparator.class.getName());
} else {
if (tezOp != null && tezOp.isSkewedJoin()) {
// TODO: PigGroupingPartitionWritableComparator only used as Group comparator in MR.
- // What should be TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_COMPARATOR_CLASS if same as MR?
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_COMPARATOR_CLASS,
+ // What should be TEZ_RUNTIME_KEY_COMPARATOR_CLASS if same as MR?
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
PigGroupingPartitionWritableComparator.class.getName());
setGroupingComparator(conf, PigGroupingPartitionWritableComparator.class.getName());
} else {
boolean hasOrderby = hasOrderby(tezOp);
conf.setClass(
- TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_COMPARATOR_CLASS,
+ TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
comparatorForKeyType(keyType, hasOrderby), RawComparator.class);
if (!hasOrderby) {
setGroupingComparator(conf, getGroupingComparatorForKeyType(keyType).getName());
@@ -1115,21 +1081,6 @@ public class TezDagBuilder extends TezOp
}
}
- void selectOutputComparator(byte keyType, Configuration conf, TezOperator tezOp)
- throws JobCreationException {
- // TODO: Handle sorting like in JobControlCompiler
- if (tezOp != null && tezOp.isSkewedJoin()) {
- // TODO: PigGroupingPartitionWritableComparator only used as Group comparator in MR.
- // What should be TEZ_RUNTIME_INTERMEDIATE_OUTPUT_KEY_COMPARATOR_CLASS if same as MR?
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_KEY_COMPARATOR_CLASS,
- PigGroupingPartitionWritableComparator.class.getName());
- } else {
- conf.setClass(
- TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_KEY_COMPARATOR_CLASS,
- comparatorForKeyType(keyType, hasOrderby(tezOp)), RawComparator.class);
- }
- }
-
private boolean hasOrderby(TezOperator tezOp) {
boolean hasOrderBy = tezOp.isGlobalSort() || tezOp.isLimitAfterSort();
if (!hasOrderBy) {
@@ -1150,7 +1101,7 @@ public class TezDagBuilder extends TezOp
// settings for the same MRJobConfig.GROUP_COMPARATOR_CLASS and use only one
conf.set(TezJobConfig.TEZ_RUNTIME_GROUP_COMPARATOR_CLASS,
comparatorClass);
- conf.set(TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_KEY_SECONDARY_COMPARATOR_CLASS,
+ conf.set(TezJobConfig.TEZ_RUNTIME_KEY_SECONDARY_COMPARATOR_CLASS,
comparatorClass);
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java?rev=1610885&r1=1610884&r2=1610885&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java Wed Jul 16 00:01:45 2014
@@ -18,7 +18,6 @@
package org.apache.pig.backend.hadoop.executionengine.tez.util;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -33,7 +32,6 @@ import org.apache.hadoop.mapreduce.lib.o
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler;
import org.apache.pig.classification.InterfaceAudience;
-import org.apache.pig.impl.util.Pair;
import org.apache.tez.common.TezJobConfig;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.mapreduce.hadoop.DeprecatedKeys;
@@ -44,34 +42,15 @@ public class MRToTezHelper {
private static final Log LOG = LogFactory.getLog(MRToTezHelper.class);
- /**
- * Keys which are used across an edge. i.e. by an Output-Input pair.
- */
- private static Map<String, Pair<String, String>> mrToTezIOParamMap = new HashMap<String, Pair<String, String>>();
-
private static List<String> mrSettingsToRetain = new ArrayList<String>();
private MRToTezHelper() {
}
static {
- populateMRToTezIOParamMap();
populateMRSettingsToRetain();
}
- private static void populateMRToTezIOParamMap() {
- mrToTezIOParamMap.put(MRJobConfig.MAP_OUTPUT_COMPRESS,
- new Pair<String, String> (
- TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_SHOULD_COMPRESS,
- TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_IS_COMPRESSED));
-
- mrToTezIOParamMap.put(MRJobConfig.MAP_OUTPUT_COMPRESS_CODEC,
- new Pair<String, String> (
- TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_OUTPUT_COMPRESS_CODEC,
- TezJobConfig.TEZ_RUNTIME_INTERMEDIATE_INPUT_COMPRESS_CODEC));
-
- }
-
private static void populateMRSettingsToRetain() {
// FileInputFormat
@@ -175,17 +154,6 @@ public class MRToTezHelper {
conf.setIfUnset(dep.getValue(), mrConf.get(dep.getKey()));
}
}
-
- for (Entry<String, Pair<String, String>> dep : mrToTezIOParamMap.entrySet()) {
- if (mrConf.get(dep.getKey()) != null) {
- conf.unset(dep.getKey());
- LOG.info("Setting " + dep.getValue() + " to "
- + mrConf.get(dep.getKey()) + " from MR setting "
- + dep.getKey());
- conf.setIfUnset(dep.getValue().first, mrConf.get(dep.getKey()));
- conf.setIfUnset(dep.getValue().second, mrConf.get(dep.getKey()));
- }
- }
}
}