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/25 06:31:31 UTC
svn commit: r1613351 - in /pig/trunk: ./
shims/test/hadoop23/org/apache/pig/test/
src/org/apache/pig/backend/hadoop/executionengine/tez/
src/org/apache/pig/backend/hadoop/executionengine/tez/util/
src/org/apache/pig/tools/pigstats/tez/
Author: rohini
Date: Fri Jul 25 04:31:30 2014
New Revision: 1613351
URL: http://svn.apache.org/r1613351
Log:
PIG-4070: Change from TezJobConfig to TezRuntimeConfiguration (rohini)
Modified:
pig/trunk/CHANGES.txt
pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java
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
pig/trunk/src/org/apache/pig/tools/pigstats/tez/TezStats.java
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1613351&r1=1613350&r2=1613351&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Fri Jul 25 04:31:30 2014
@@ -46,6 +46,8 @@ OPTIMIZATIONS
BUG FIXES
+PIG-4070: Change from TezJobConfig to TezRuntimeConfiguration (rohini)
+
PIG-4068: ObjectCache causes ClassCastException (cheolsoo)
PIG-4067: TestAllLoader in piggybank fails with new hive version (rohini)
Modified: pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java
URL: http://svn.apache.org/viewvc/pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java?rev=1613351&r1=1613350&r2=1613351&view=diff
==============================================================================
--- pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java (original)
+++ pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java Fri Jul 25 04:31:30 2014
@@ -33,7 +33,7 @@ import org.apache.pig.ExecType;
import org.apache.pig.PigConfiguration;
import org.apache.pig.backend.hadoop.executionengine.tez.TezExecType;
import org.apache.pig.backend.hadoop.executionengine.tez.TezSessionManager;
-import org.apache.tez.common.TezJobConfig;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
public class TezMiniCluster extends MiniGenericCluster {
private static final File CONF_DIR = new File("build/classes");
@@ -106,7 +106,7 @@ public class TezMiniCluster extends Mini
// Write tez-site.xml
Configuration tez_conf = new Configuration(false);
// TODO PIG-3659 - Remove this once memory management is fixed
- tez_conf.set(TezJobConfig.TEZ_RUNTIME_IO_SORT_MB, "20");
+ tez_conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, "20");
tez_conf.set("tez.lib.uris", "hdfs:///tez,hdfs:///tez/lib");
tez_conf.writeXml(new FileOutputStream(TEZ_CONF_FILE));
m_fileSys.copyFromLocalFile(
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=1613351&r1=1613350&r2=1613351&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 Fri Jul 25 04:31:30 2014
@@ -105,7 +105,6 @@ import org.apache.pig.impl.plan.Operator
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.ObjectSerializer;
import org.apache.pig.impl.util.UDFContext;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezUtils;
import org.apache.tez.dag.api.DAG;
import org.apache.tez.dag.api.Edge;
@@ -114,6 +113,8 @@ import org.apache.tez.dag.api.EdgeProper
import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.dag.api.GroupInputEdge;
import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.InputInitializerDescriptor;
+import org.apache.tez.dag.api.OutputCommitterDescriptor;
import org.apache.tez.dag.api.OutputDescriptor;
import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.Vertex;
@@ -129,6 +130,7 @@ import org.apache.tez.mapreduce.hadoop.M
import org.apache.tez.mapreduce.input.MRInput;
import org.apache.tez.mapreduce.output.MROutput;
import org.apache.tez.mapreduce.partition.MRPartitioner;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.input.ConcatenatedMergedKeyValueInput;
import org.apache.tez.runtime.library.input.ShuffledMergedInput;
import org.apache.tez.runtime.library.input.SortedGroupedMergedInput;
@@ -230,9 +232,9 @@ public class TezDagBuilder extends TezOp
tezOp.getVertexGroupInfo().setVertexGroup(vertexGroup);
POStore store = tezOp.getVertexGroupInfo().getStore();
if (store != null) {
- vertexGroup.addOutput(store.getOperatorKey().toString(),
+ vertexGroup.addDataSink(store.getOperatorKey().toString(),
tezOp.getVertexGroupInfo().getStoreOutputDescriptor(),
- MROutputCommitter.class);
+ new OutputCommitterDescriptor(MROutputCommitter.class.getName()));
}
}
}
@@ -290,21 +292,21 @@ public class TezDagBuilder extends TezOp
}
}
- conf.setIfUnset(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS,
+ conf.setIfUnset(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS,
MRPartitioner.class.getName());
if (edge.getIntermediateOutputKeyClass() != null) {
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS,
edge.getIntermediateOutputKeyClass());
}
if (edge.getIntermediateOutputValueClass() != null) {
- conf.set(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS,
edge.getIntermediateOutputValueClass());
}
if (edge.getIntermediateOutputKeyComparatorClass() != null) {
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
edge.getIntermediateOutputKeyComparatorClass());
}
@@ -327,10 +329,10 @@ public class TezDagBuilder extends TezOp
// 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
// In MR - job.setSortComparatorClass() or MRJobConfig.KEY_COMPARATOR
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
PigSecondaryKeyComparator.class.getName());
// In MR - job.setOutputKeyClass() or MRJobConfig.OUTPUT_KEY_CLASS
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, NullableTuple.class.getName());
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, NullableTuple.class.getName());
setGroupingComparator(conf, PigSecondaryKeyGroupComparator.class.getName());
}
@@ -369,7 +371,7 @@ public class TezDagBuilder extends TezOp
lrDiscoverer.visit();
combinePlan.remove(combPack);
- conf.set(TezJobConfig.TEZ_RUNTIME_COMBINER_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS,
MRCombiner.class.getName());
conf.set(MRJobConfig.COMBINE_CLASS_ATTR,
PigCombiner.Combine.class.getName());
@@ -680,12 +682,12 @@ public class TezDagBuilder extends TezOp
// TODO: These should get the globalConf, or a merged version that
// keeps settings like pig.maxCombinedSplitSize
vertex.setTaskLocationsHint(inputSplitInfo.getTaskLocationHints());
- vertex.addInput(ld.getOperatorKey().toString(),
+ vertex.addDataSource(ld.getOperatorKey().toString(),
new InputDescriptor(MRInput.class.getName())
.setUserPayload(MRHelpers.createMRInputPayload(
userPayload,
inputSplitInfo.getSplitsProto())),
- MRInputSplitDistributor.class);
+ new InputInitializerDescriptor(MRInputSplitDistributor.class.getName()));
}
for (POStore store : stores) {
@@ -711,8 +713,8 @@ public class TezDagBuilder extends TezOp
continue;
}
}
- vertex.addOutput(store.getOperatorKey().toString(),
- storeOutDescriptor, MROutputCommitter.class);
+ vertex.addDataSink(store.getOperatorKey().toString(),
+ storeOutDescriptor, new OutputCommitterDescriptor(MROutputCommitter.class.getName()));
}
// LoadFunc and StoreFunc add delegation tokens to Job Credentials in
@@ -916,21 +918,21 @@ public class TezDagBuilder extends 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_KEY_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS,
NullableTuple.class.getName());
} else if (tezOp != null && tezOp.isSkewedJoin() && isConnectedToPackage) {
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS,
NullablePartitionWritable.class.getName());
} else {
Class<? extends WritableComparable> keyClass = HDataType
.getWritableComparableTypes(keyType).getClass();
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS,
keyClass.getName());
}
- conf.set(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS,
NullableTuple.class.getName());
- conf.set(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS,
MRPartitioner.class.getName());
selectOutputComparator(keyType, conf, tezOp);
}
@@ -1059,20 +1061,20 @@ public class TezDagBuilder extends TezOp
// TODO: Handle sorting like in JobControlCompiler
// TODO: Group comparators as in JobControlCompiler
if (tezOp != null && tezOp.isUseSecondaryKey()) {
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
+ conf.set(TezRuntimeConfiguration.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_KEY_COMPARATOR_CLASS if same as MR?
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
PigGroupingPartitionWritableComparator.class.getName());
setGroupingComparator(conf, PigGroupingPartitionWritableComparator.class.getName());
} else {
boolean hasOrderby = hasOrderby(tezOp);
conf.setClass(
- TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
+ TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
comparatorForKeyType(keyType, hasOrderby), RawComparator.class);
if (!hasOrderby) {
setGroupingComparator(conf, getGroupingComparatorForKeyType(keyType).getName());
@@ -1099,9 +1101,9 @@ public class TezDagBuilder extends TezOp
// In MR - job.setGroupingComparatorClass() or MRJobConfig.GROUP_COMPARATOR_CLASS
// TODO: Check why tez-mapreduce ReduceProcessor use two different tez
// settings for the same MRJobConfig.GROUP_COMPARATOR_CLASS and use only one
- conf.set(TezJobConfig.TEZ_RUNTIME_GROUP_COMPARATOR_CLASS,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_GROUP_COMPARATOR_CLASS,
comparatorClass);
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_SECONDARY_COMPARATOR_CLASS,
+ conf.set(TezRuntimeConfiguration.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=1613351&r1=1613350&r2=1613351&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 Fri Jul 25 04:31:30 2014
@@ -32,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.tez.common.TezJobConfig;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.mapreduce.hadoop.DeprecatedKeys;
import org.apache.tez.mapreduce.hadoop.MRJobConfig;
@@ -122,7 +121,7 @@ public class MRToTezHelper {
MRJobConfig.DEFAULT_MR_AM_MAX_ATTEMPTS));
if (tezConf.get("mapreduce.job.credentials.binary") != null) {
- dagAMConf.setIfUnset(TezJobConfig.TEZ_CREDENTIALS_PATH,
+ dagAMConf.setIfUnset(TezConfiguration.TEZ_CREDENTIALS_PATH,
tezConf.get("mapreduce.job.credentials.binary"));
}
Modified: pig/trunk/src/org/apache/pig/tools/pigstats/tez/TezStats.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/tools/pigstats/tez/TezStats.java?rev=1613351&r1=1613350&r2=1613351&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/tools/pigstats/tez/TezStats.java (original)
+++ pig/trunk/src/org/apache/pig/tools/pigstats/tez/TezStats.java Fri Jul 25 04:31:30 2014
@@ -188,6 +188,8 @@ public class TezStats extends PigStats {
*/
public void accumulateStats(TezJob tezJob) throws IOException {
DAGStatus dagStatus = tezJob.getDAGStatus();
+ if (dagStatus == null)
+ return;
if (dagStatus.getState() == DAGStatus.State.SUCCEEDED) {
addVertexStats(tezJob, true);
dagStatsStrings.add(getDisplayString(tezJob));