You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ch...@apache.org on 2013/12/18 19:51:12 UTC
svn commit: r1552058 - in /pig/branches/tez:
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/
src/org/apache/pig/backend/hadoop/executionengine/tez/ test/e2e/pig/tests/
test/org/apache/pig/test/data/GoldenFiles/ test...
Author: cheolsoo
Date: Wed Dec 18 18:51:11 2013
New Revision: 1552058
URL: http://svn.apache.org/r1552058
Log:
PIG-3604: Implement replicated join in Tez (cheolsoo)
Added:
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POFRJoinTez.java
pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC10.gld
pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC11.gld
Modified:
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POFRJoin.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POMergeJoin.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POLocalRearrangeTez.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POShuffleTezLoad.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POUnionTezLoad.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java
pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java
pig/branches/tez/test/e2e/pig/tests/tez.conf
pig/branches/tez/test/org/apache/pig/tez/TestTezCompiler.java
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POFRJoin.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POFRJoin.java?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POFRJoin.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POFRJoin.java Wed Dec 18 18:51:11 2013
@@ -67,48 +67,44 @@ import org.apache.pig.impl.plan.VisitorE
*/
// We intentionally skip type checking in backend for performance reasons
-@SuppressWarnings("unchecked")
public class POFRJoin extends PhysicalOperator {
private static final Log log = LogFactory.getLog(POFRJoin.class);
- /**
- *
- */
private static final long serialVersionUID = 1L;
// The number in the input list which denotes the fragmented input
- private int fragment;
+ protected int fragment;
// There can be n inputs each being a List<PhysicalPlan>
// Ex. join A by ($0+$1,$0-$1), B by ($0*$1,$0/$1);
- private List<List<PhysicalPlan>> phyPlanLists;
+ protected List<List<PhysicalPlan>> phyPlanLists;
// The key type for each Local Rearrange operator
- private List<List<Byte>> keyTypes;
+ protected List<List<Byte>> keyTypes;
// The Local Rearrange operators modeling the join key
- private POLocalRearrange[] LRs;
+ protected POLocalRearrange[] LRs;
// The set of files that represent the replicated inputs
- private FileSpec[] replFiles;
+ protected FileSpec[] replFiles;
// Used to configure the foreach operator
- private ConstantExpression[] constExps;
+ protected ConstantExpression[] constExps;
// Used to produce the cross product of various bags
- private POForEach fe;
+ protected POForEach fe;
// The array of Hashtables one per replicated input. replicates[fragment] =
// null
// fragment is the input which is fragmented and not replicated.
- private TupleToMapKey replicates[];
+ protected TupleToMapKey replicates[];
// varaible which denotes whether we are returning tuples from the foreach
// operator
- private boolean processingPlan;
+ protected boolean processingPlan;
// A dummy tuple
- private Tuple dumTup = TupleFactory.getInstance().newTuple(1);
+ protected Tuple dumTup = TupleFactory.getInstance().newTuple(1);
// An instance of tuple factory
- private transient TupleFactory mTupleFactory;
- private boolean setUp;
+ protected transient TupleFactory mTupleFactory;
+ protected boolean setUp;
// A Boolean variable which denotes if this is a LeftOuter Join or an Inner
// Join
- private boolean isLeftOuterJoin;
+ protected boolean isLeftOuterJoin;
- // This list contains nullTuples according to schema of various inputs
- private DataBag nullBag;
- private Schema[] inputSchemas;
- private Schema[] keySchemas;
+ // This list contains nullTuples according to schema of various inputs
+ protected DataBag nullBag;
+ protected Schema[] inputSchemas;
+ protected Schema[] keySchemas;
public POFRJoin(OperatorKey k, int rp, List<PhysicalOperator> inp,
List<List<PhysicalPlan>> ppLists, List<List<Byte>> keyTypes,
@@ -152,8 +148,22 @@ public class POFRJoin extends PhysicalOp
}
}
- public List<List<PhysicalPlan>> getJoinPlans() {
- return phyPlanLists;
+ public POFRJoin(POFRJoin copy) throws ExecException {
+ super(copy);
+ this.phyPlanLists = copy.phyPlanLists;
+ this.fragment = copy.fragment;
+ this.keyTypes = copy.keyTypes;
+ this.replFiles = copy.replFiles;
+ this.replicates = copy.replicates;
+ this.LRs = copy.LRs;
+ this.fe = copy.fe;
+ this.constExps = copy.constExps;
+ this.processingPlan = copy.processingPlan;
+ this.mTupleFactory = copy.mTupleFactory;
+ this.nullBag = copy.nullBag;
+ this.isLeftOuterJoin = copy.isLeftOuterJoin;
+ this.inputSchemas = copy.inputSchemas;
+ this.keySchemas = copy.keySchemas;
}
private OperatorKey genKey(OperatorKey old) {
@@ -259,9 +269,9 @@ public class POFRJoin extends PhysicalOp
// Process the current input
inp = processInput();
if (inp.returnStatus == POStatus.STATUS_EOP
- || inp.returnStatus == POStatus.STATUS_ERR)
+ || inp.returnStatus == POStatus.STATUS_ERR) {
return inp;
- if (inp.returnStatus == POStatus.STATUS_NULL) {
+ } else if (inp.returnStatus == POStatus.STATUS_NULL) {
continue;
}
@@ -301,8 +311,9 @@ public class POFRJoin extends PhysicalOp
// If this is not LeftOuter Join and there was no match we
// skip the processing of this left tuple and move ahead
- if (!isLeftOuterJoin && noMatch)
+ if (!isLeftOuterJoin && noMatch) {
continue;
+ }
fe.attachInput(dumTup);
processingPlan = true;
@@ -318,7 +329,7 @@ public class POFRJoin extends PhysicalOp
}
}
- private static class TupleToMapKey {
+ protected static class TupleToMapKey {
private HashMap<Tuple, TuplesToSchemaTupleList> tuples;
private SchemaTupleFactory tf;
@@ -345,10 +356,10 @@ public class POFRJoin extends PhysicalOp
/**
* Builds the HashMaps by reading each replicated input from the DFS using a
* Load operator
- *
+ *
* @throws ExecException
*/
- private void setUpHashMap() throws ExecException {
+ protected void setUpHashMap() throws ExecException {
SchemaTupleFactory[] inputSchemaTupleFactories = new SchemaTupleFactory[inputSchemas.length];
SchemaTupleFactory[] keySchemaTupleFactories = new SchemaTupleFactory[inputSchemas.length];
for (int i = 0; i < inputSchemas.length; i++) {
@@ -416,7 +427,7 @@ public class POFRJoin extends PhysicalOp
log.debug("Hash Table built. Time taken: " + (time2 - time1));
}
- private boolean isKeyNull(Object key) throws ExecException {
+ protected boolean isKeyNull(Object key) throws ExecException {
if (key == null) return true;
if (key instanceof Tuple) {
Tuple t = (Tuple)key;
@@ -437,7 +448,7 @@ public class POFRJoin extends PhysicalOp
/*
* Extracts the value tuple from the LR operator's output tuple
*/
- private Tuple getValueTuple(POLocalRearrange lr, Tuple tuple)
+ protected Tuple getValueTuple(POLocalRearrange lr, Tuple tuple)
throws ExecException {
Tuple val = (Tuple) tuple.get(2);
Tuple retTup = null;
@@ -491,6 +502,14 @@ public class POFRJoin extends PhysicalOp
return retTup;
}
+ public List<List<PhysicalPlan>> getJoinPlans() {
+ return phyPlanLists;
+ }
+
+ public POLocalRearrange[] getLRs() {
+ return LRs;
+ }
+
public int getFragment() {
return fragment;
}
@@ -506,7 +525,7 @@ public class POFRJoin extends PhysicalOp
public void setReplFiles(FileSpec[] replFiles) {
this.replFiles = replFiles;
}
-
+
@Override
public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
// no op: all handled by the preceding POForEach
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POMergeJoin.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POMergeJoin.java?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POMergeJoin.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POMergeJoin.java Wed Dec 18 18:51:11 2013
@@ -223,11 +223,11 @@ public class POMergeJoin extends Physica
* from Tuple to SchemaTuple. This is necessary because we are not getting SchemaTuples
* from the source, though in the future that is what we would like to do.
*/
- protected static class TuplesToSchemaTupleList {
+ public static class TuplesToSchemaTupleList {
private List<Tuple> tuples;
private SchemaTupleFactory tf;
- protected TuplesToSchemaTupleList(int ct, TupleMaker<?> tf) {
+ public TuplesToSchemaTupleList(int ct, TupleMaker<?> tf) {
tuples = new ArrayList<Tuple>(ct);
if (tf instanceof SchemaTupleFactory) {
this.tf = (SchemaTupleFactory)tf;
Added: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POFRJoinTez.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POFRJoinTez.java?rev=1552058&view=auto
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POFRJoinTez.java (added)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POFRJoinTez.java Wed Dec 18 18:51:11 2013
@@ -0,0 +1,170 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.backend.hadoop.executionengine.tez;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFRJoin;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeJoin.TuplesToSchemaTupleList;
+import org.apache.pig.data.SchemaTupleBackend;
+import org.apache.pig.data.SchemaTupleClassGenerator.GenContext;
+import org.apache.pig.data.SchemaTupleFactory;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.io.NullableTuple;
+import org.apache.pig.impl.io.PigNullableWritable;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.tez.runtime.api.LogicalInput;
+import org.apache.tez.runtime.library.broadcast.input.BroadcastKVReader;
+import org.apache.tez.runtime.library.input.ShuffledUnorderedKVInput;
+
+import com.google.common.collect.Lists;
+
+/**
+ * POFRJoinTez is used on the backend to load replicated table from Tez
+ * ShuffleUnorderedKVInput and load fragmented table from data pipeline.
+ */
+public class POFRJoinTez extends POFRJoin implements TezLoad {
+
+ private static final Log log = LogFactory.getLog(POFRJoinTez.class);
+ private static final long serialVersionUID = 1L;
+
+ // For replicated tables
+ private List<ShuffledUnorderedKVInput> replInputs = Lists.newArrayList();
+ @SuppressWarnings("rawtypes")
+ private List<BroadcastKVReader> replReaders = Lists.newArrayList();
+ private List<String> inputKeys;
+ private int currIdx = 0;
+
+ public POFRJoinTez(POFRJoin copy, List<String> inputKeys) throws ExecException {
+ super(copy);
+ this.inputKeys = inputKeys;
+ }
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ public void attachInputs(Map<String, LogicalInput> inputs, Configuration conf)
+ throws ExecException {
+ try {
+ for (String key : inputKeys) {
+ LogicalInput input = inputs.get(key);
+ if (input instanceof ShuffledUnorderedKVInput) {
+ ShuffledUnorderedKVInput suInput = (ShuffledUnorderedKVInput) input;
+ this.replInputs.add(suInput);
+ this.replReaders.add((BroadcastKVReader) suInput.getReader());
+ }
+ }
+ } catch (Exception e) {
+ throw new ExecException(e);
+ }
+ }
+
+ /**
+ * Builds the HashMaps by reading replicated inputs from broadcast edges
+ *
+ * @throws ExecException
+ */
+ @Override
+ protected void setUpHashMap() throws ExecException {
+ SchemaTupleFactory[] inputSchemaTupleFactories = new SchemaTupleFactory[inputSchemas.length];
+ SchemaTupleFactory[] keySchemaTupleFactories = new SchemaTupleFactory[inputSchemas.length];
+
+ for (int i = 0; i < inputSchemas.length; i++) {
+ Schema schema = inputSchemas[i];
+ if (schema != null) {
+ log.debug("Using SchemaTuple for FR Join Schema: " + schema);
+ inputSchemaTupleFactories[i] =
+ SchemaTupleBackend.newSchemaTupleFactory(schema, false, GenContext.FR_JOIN);
+ }
+ schema = keySchemas[i];
+ if (schema != null) {
+ log.debug("Using SchemaTuple for FR Join key Schema: " + schema);
+ keySchemaTupleFactories[i] =
+ SchemaTupleBackend.newSchemaTupleFactory(schema, false, GenContext.FR_JOIN);
+ }
+ }
+
+ long time1 = System.currentTimeMillis();
+ log.debug("Completed setup. Trying to build replication hash table");
+
+ replicates[fragment] = null;
+ while (currIdx < replInputs.size()) {
+ // We need to adjust the index because the number of replInputs is
+ // one less than the number of inputSchemas. The inputSchemas
+ // includes the fragmented table.
+ int adjustedIdx = currIdx == fragment ? currIdx + 1 : currIdx;
+ SchemaTupleFactory inputSchemaTupleFactory = inputSchemaTupleFactories[adjustedIdx];
+ SchemaTupleFactory keySchemaTupleFactory = keySchemaTupleFactories[adjustedIdx];
+
+ TupleToMapKey replicate = new TupleToMapKey(1000, keySchemaTupleFactory);
+ POLocalRearrange lr = LRs[adjustedIdx];
+
+ try {
+ while (replReaders.get(currIdx).next()) {
+ if (getReporter() != null) {
+ getReporter().progress();
+ }
+
+ PigNullableWritable key = (PigNullableWritable) replReaders.get(currIdx).getCurrentKey();
+ if (isKeyNull(key.getValueAsPigType())) continue;
+ NullableTuple val = (NullableTuple) replReaders.get(currIdx).getCurrentValue();
+
+ // POFRJoin#getValueTuple() is reused to construct valTuple,
+ // and it takes an indexed Tuple as parameter. So we need to
+ // construct one here.
+ Tuple retTuple = mTupleFactory.newTuple(3);
+ retTuple.set(0, key.getIndex());
+ retTuple.set(1, key.getValueAsPigType());
+ retTuple.set(2, val.getValueAsPigType());
+ Tuple valTuple = getValueTuple(lr, retTuple);
+
+ Tuple keyTuple = mTupleFactory.newTuple(1);
+ keyTuple.set(0, key.getValueAsPigType());
+ if (replicate.get(keyTuple) == null) {
+ replicate.put(keyTuple, new TuplesToSchemaTupleList(1, inputSchemaTupleFactory));
+ }
+ replicate.get(keyTuple).add(valTuple);
+ }
+ } catch (IOException e) {
+ throw new ExecException(e);
+ }
+ replicates[adjustedIdx] = replicate;
+ currIdx++;
+ }
+
+ long time2 = System.currentTimeMillis();
+ log.info("Hash Table built. Time taken: " + (time2 - time1));
+ }
+
+ @Override
+ public String name() {
+ StringBuffer inputs = new StringBuffer();
+ for (int i = 0; i < inputKeys.size(); i++) {
+ if (i > 0) inputs.append(",");
+ inputs.append(inputKeys.get(i));
+ }
+ return super.name() + "\t<-\t " + inputs.toString();
+ }
+}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POLocalRearrangeTez.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POLocalRearrangeTez.java?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POLocalRearrangeTez.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POLocalRearrangeTez.java Wed Dec 18 18:51:11 2013
@@ -28,12 +28,10 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
import org.apache.pig.data.Tuple;
-import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.io.NullableTuple;
import org.apache.pig.impl.io.PigNullableWritable;
import org.apache.pig.impl.plan.NodeIdGenerator;
import org.apache.pig.impl.plan.OperatorKey;
-import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.runtime.api.LogicalOutput;
import org.apache.tez.runtime.library.api.KeyValueWriter;
import org.apache.tez.runtime.library.output.OnFileSortedOutput;
@@ -46,10 +44,9 @@ import org.apache.tez.runtime.library.ou
public class POLocalRearrangeTez extends POLocalRearrange implements TezOutput {
private static final long serialVersionUID = 1L;
- private static TupleFactory tupleFactory = TupleFactory.getInstance();
private static Result empty = new Result(POStatus.STATUS_NULL, null);
+
private String outputKey;
- private DataMovementType outputType;
private transient KeyValueWriter writer;
// Tez union is implemented as LR + Pkg
private boolean isUnion = false;
@@ -96,10 +93,8 @@ public class POLocalRearrangeTez extends
try {
if (logicalOut instanceof OnFileSortedOutput) {
writer = ((OnFileSortedOutput) logicalOut).getWriter();
- outputType = DataMovementType.SCATTER_GATHER;
} else if (logicalOut instanceof OnFileUnorderedKVOutput) {
writer = ((OnFileUnorderedKVOutput) logicalOut).getWriter();
- outputType = DataMovementType.BROADCAST;
} else {
throw new ExecException(
"POLocalRearrangeTez only accepts OnFileSortedOutput (shuffle)"
@@ -117,57 +112,38 @@ public class POLocalRearrangeTez extends
@Override
public Result getNextTuple() throws ExecException {
Result res = super.getNextTuple();
- if (outputType == null) { //In the case of combiner
+ if (writer == null) { // In the case of combiner
return res;
}
+
try {
switch (res.returnStatus) {
case POStatus.STATUS_OK:
if (illustrator == null) {
Tuple result = (Tuple) res.result;
- switch (outputType) {
- case SCATTER_GATHER: {
- Byte index = (Byte)result.get(0);
- PigNullableWritable key = null;
- NullableTuple val = null;
- if (isUnion()) {
- // Use the entire tuple as both key and value
- key = HDataType.getWritableComparableTypes(result.get(1), keyType);
- val = new NullableTuple((Tuple)result.get(1));
- } else {
- key = HDataType.getWritableComparableTypes(result.get(1), keyType);
- val = new NullableTuple((Tuple)result.get(2));
- }
-
- // Both the key and the value need the index. The key needs it so
- // that it can be sorted on the index in addition to the key
- // value. The value needs it so that POPackage can properly
- // assign the tuple to its slot in the projection.
- key.setIndex(index);
- val.setIndex(index);
- writer.write(key, val);
- break;
- }
- case BROADCAST: {
- Byte index = (Byte)result.get(0);
- // Key isn't used by broadcast edge, so it can be an empty tuple.
- Tuple nullTuple = tupleFactory.newTuple();
- PigNullableWritable key =
- HDataType.getWritableComparableTypes(nullTuple, keyType);
- NullableTuple val = new NullableTuple((Tuple)result.get(1));
- key.setIndex(index);
- val.setIndex(index);
- writer.write(key, val);
- break;
- }
- default:
- throw new IOException("Unkonwn output type: " + outputType);
+ Byte index = (Byte)result.get(0);
+ PigNullableWritable key = null;
+ NullableTuple val = null;
+ if (isUnion()) {
+ // Use the entire tuple as both key and value
+ key = HDataType.getWritableComparableTypes(result.get(1), keyType);
+ val = new NullableTuple((Tuple)result.get(1));
+ } else {
+ key = HDataType.getWritableComparableTypes(result.get(1), keyType);
+ val = new NullableTuple((Tuple)result.get(2));
}
+
+ // Both the key and the value need the index. The key needs it so
+ // that it can be sorted on the index in addition to the key
+ // value. The value needs it so that POPackage can properly
+ // assign the tuple to its slot in the projection.
+ key.setIndex(index);
+ val.setIndex(index);
+ writer.write(key, val);
} else {
illustratorMarkup(res.result, res.result, 0);
}
res = empty;
-
break;
case POStatus.STATUS_EOP:
case POStatus.STATUS_ERR:
@@ -180,6 +156,7 @@ public class POLocalRearrangeTez extends
String msg = "Received error from POLocalRearrage function." + ioe.getMessage();
throw new ExecException(msg, errCode, ioe);
}
+
return res;
}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POShuffleTezLoad.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POShuffleTezLoad.java?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POShuffleTezLoad.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POShuffleTezLoad.java Wed Dec 18 18:51:11 2013
@@ -34,7 +34,6 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.data.DataBag;
import org.apache.pig.data.InternalCachedBag;
import org.apache.pig.data.Tuple;
-import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.io.NullableTuple;
import org.apache.pig.impl.io.PigNullableWritable;
import org.apache.pig.impl.plan.OperatorKey;
@@ -46,19 +45,14 @@ public class POShuffleTezLoad extends PO
private static final long serialVersionUID = 1L;
- List<String> inputKeys = new ArrayList<String>();
-
- List<ShuffledMergedInput> inputs = new ArrayList<ShuffledMergedInput>();
- List<KeyValuesReader> readers = new ArrayList<KeyValuesReader>();
- List<Boolean> finished = new ArrayList<Boolean>();
+ protected List<String> inputKeys = new ArrayList<String>();
+ protected List<ShuffledMergedInput> inputs = new ArrayList<ShuffledMergedInput>();
+ protected List<KeyValuesReader> readers = new ArrayList<KeyValuesReader>();
+ private boolean[] finished;
private boolean[] readOnce;
- Result res;
-
- protected static final TupleFactory tf = TupleFactory.getInstance();
-
- WritableComparator comparator = null;
+ private WritableComparator comparator = null;
public POShuffleTezLoad(OperatorKey k, POPackage pack) {
super(k);
@@ -69,11 +63,6 @@ public class POShuffleTezLoad extends PO
@Override
public void attachInputs(Map<String, LogicalInput> inputs, Configuration conf)
throws ExecException {
- readOnce = new boolean[numInputs];
- for (int i = 0; i < numInputs; i++) {
- readOnce[i] = false;
- }
-
try {
comparator = ReflectionUtils.newInstance(
TezDagBuilder.comparatorForKeyType(pkgr.getKeyType()), conf);
@@ -81,15 +70,29 @@ public class POShuffleTezLoad extends PO
throw new ExecException(e);
}
try {
- // TODO: Only take the inputs which are actually specified.
- for (LogicalInput input : inputs.values()) {
- ShuffledMergedInput smInput = (ShuffledMergedInput) input;
- this.inputs.add(smInput);
- this.readers.add(smInput.getReader());
+ for (String key : inputKeys) {
+ LogicalInput input = inputs.get(key);
+ if (input instanceof ShuffledMergedInput) {
+ ShuffledMergedInput smInput = (ShuffledMergedInput) input;
+ this.inputs.add(smInput);
+ this.readers.add(smInput.getReader());
+ }
+ }
+
+ // We need to adjust numInputs because it's possible for both
+ // ShuffledMergedInputs and non-ShuffledMergedInputs to be attached
+ // to the same vertex. If so, we're only interested in
+ // ShuffledMergedInputs. So we ignore the others.
+ this.numInputs = this.inputs.size();
+
+ readOnce = new boolean[numInputs];
+ for (int i = 0; i < numInputs; i++) {
+ readOnce[i] = false;
}
+ finished = new boolean[numInputs];
for (int i = 0; i < numInputs; i++) {
- finished.add(!readers.get(i).next());
+ finished[i] = !readers.get(i).next();
}
} catch (IOException e) {
throw new ExecException(e);
@@ -104,7 +107,7 @@ public class POShuffleTezLoad extends PO
boolean newData = false;
try {
for (int i = 0; i < numInputs; i++) {
- if (!finished.get(i)) {
+ if (!finished[i]) {
newData = true;
PigNullableWritable current = (PigNullableWritable) readers
.get(i).getCurrentKey();
@@ -128,7 +131,7 @@ public class POShuffleTezLoad extends PO
try {
for (int i = 0; i < numInputs; i++) {
- if (!finished.get(i)) {
+ if (!finished[i]) {
PigNullableWritable current = (PigNullableWritable) readers
.get(i).getCurrentKey();
if (comparator.compare(minimum, current) == 0) {
@@ -140,7 +143,7 @@ public class POShuffleTezLoad extends PO
Tuple tup = pkgr.getValueTuple(key, nTup, index);
bag.add(tup);
}
- finished.set(i, !readers.get(i).next());
+ finished[i] = !readers.get(i).next();
bags[i] = bag;
} else {
bags[i] = new InternalCachedBag(numInputs);
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POUnionTezLoad.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POUnionTezLoad.java?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POUnionTezLoad.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/POUnionTezLoad.java Wed Dec 18 18:51:11 2013
@@ -42,6 +42,7 @@ public class POUnionTezLoad extends POSh
private static Result eopResult = new Result(POStatus.STATUS_EOP, null);
private int currIdx = 0;
private int tuplCnt = 0;
+ private Result res;
public POUnionTezLoad(OperatorKey k, POPackage pack) {
super(k, pack);
@@ -51,11 +52,20 @@ public class POUnionTezLoad extends POSh
public void attachInputs(Map<String, LogicalInput> inputs, Configuration conf)
throws ExecException {
try {
- for (LogicalInput input : inputs.values()) {
- ShuffledMergedInput suInput = (ShuffledMergedInput) input;
- this.inputs.add(suInput);
- this.readers.add((KeyValuesReader) suInput.getReader());
+ for (String key : inputKeys) {
+ LogicalInput input = inputs.get(key);
+ if (input instanceof ShuffledMergedInput) {
+ ShuffledMergedInput suInput = (ShuffledMergedInput) input;
+ this.inputs.add(suInput);
+ this.readers.add((KeyValuesReader) suInput.getReader());
+ }
}
+
+ // We need to adjust numInputs because it's possible for both
+ // ShuffledMergedInputs and non-ShuffledMergedInputs to be attached
+ // to the same vertex. If so, we're only interested in
+ // ShuffledMergedInputs. So we ignore the others.
+ this.numInputs = this.inputs.size();
} catch (Exception e) {
throw new ExecException(e);
}
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/PigProcessor.java Wed Dec 18 18:51:11 2013
@@ -121,6 +121,10 @@ public class PigProcessor implements Log
for (POShuffleTezLoad shuffle : shuffles){
shuffle.attachInputs(inputs, conf);
}
+ LinkedList<POFRJoinTez> broadcasts = PlanHelper.getPhysicalOperators(execPlan, POFRJoinTez.class);
+ for (POFRJoinTez broadcast : broadcasts){
+ broadcast.attachInputs(inputs, conf);
+ }
}
private void initializeOutputs(Map<String, LogicalOutput> outputs) throws Exception {
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezCompiler.java Wed Dec 18 18:51:11 2013
@@ -81,6 +81,9 @@ import org.apache.pig.impl.plan.PlanExce
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.Pair;
import org.apache.pig.impl.util.Utils;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.runtime.library.input.ShuffledUnorderedKVInput;
+import org.apache.tez.runtime.library.output.OnFileUnorderedKVOutput;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -713,9 +716,69 @@ public class TezCompiler extends PhyPlan
@Override
public void visitFRJoin(POFRJoin op) throws VisitorException {
- int errCode = 2034;
- String msg = "Cannot compile " + op.getClass().getSimpleName();
- throw new TezCompilerException(msg, errCode, PigException.BUG);
+ try {
+ FileSpec[] replFiles = new FileSpec[op.getInputs().size()];
+ for (int i = 0; i < replFiles.length; i++) {
+ if (i == op.getFragment()) {
+ continue;
+ }
+ replFiles[i] = getTempFileSpec();
+ }
+ op.setReplFiles(replFiles);
+
+ List<String> inputKeys = Lists.newArrayList();
+ curTezOp = phyToTezOpMap.get(op.getInputs().get(op.getFragment()));
+
+ for (int i = 0; i < compiledInputs.length; i++) {
+ TezOperator tezOp = compiledInputs[i];
+ if (curTezOp.equals(tezOp)) {
+ continue;
+ }
+
+ if (!tezOp.isClosed()) {
+ POLocalRearrangeTez lr = new POLocalRearrangeTez(op.getLRs()[i]);
+ lr.setOutputKey(curTezOp.getOperatorKey().toString());
+
+ tezOp.plan.addAsLeaf(lr);
+ connect(tezPlan, tezOp, curTezOp);
+ inputKeys.add(tezOp.getOperatorKey().toString());
+
+ // Configure broadcast edges for replicated tables
+ TezEdgeDescriptor edge = curTezOp.inEdges.get(tezOp.getOperatorKey());
+ edge.dataMovementType = DataMovementType.BROADCAST;
+ edge.outputClassName = OnFileUnorderedKVOutput.class.getName();
+ edge.inputClassName = ShuffledUnorderedKVInput.class.getName();
+ } else {
+ int errCode = 2022;
+ String msg = "The current operator is closed. This is unexpected while compiling.";
+ throw new TezCompilerException(msg, errCode, PigException.BUG);
+ }
+ }
+
+ if (!curTezOp.isClosed()) {
+ curTezOp.plan.addAsLeaf(new POFRJoinTez(op, inputKeys));
+ } else {
+ int errCode = 2022;
+ String msg = "The current operator is closed. This is unexpected while compiling.";
+ throw new TezCompilerException(msg, errCode, PigException.BUG);
+ }
+
+ List<List<PhysicalPlan>> joinPlans = op.getJoinPlans();
+ if (joinPlans != null) {
+ for (List<PhysicalPlan> joinPlan : joinPlans) {
+ if (joinPlan != null) {
+ for (PhysicalPlan plan : joinPlan) {
+ processUDFs(plan);
+ }
+ }
+ }
+ }
+ phyToTezOpMap.put(op, curTezOp);
+ } catch (Exception e) {
+ int errCode = 2034;
+ String msg = "Error compiling operator " + op.getClass().getSimpleName();
+ throw new TezCompilerException(msg, errCode, PigException.BUG, e);
+ }
}
@Override
Modified: pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java (original)
+++ pig/branches/tez/src/org/apache/pig/backend/hadoop/executionengine/tez/TezDagBuilder.java Wed Dec 18 18:51:11 2013
@@ -277,7 +277,6 @@ public class TezDagBuilder extends TezOp
payloadConf.setClass("mapreduce.inputformat.class",
PigInputFormat.class, InputFormat.class);
-
// Configure the classes for incoming shuffles to this TezOp
List<PhysicalOperator> roots = tezOp.plan.getRoots();
if (roots.size() == 1 && roots.get(0) instanceof POPackage) {
@@ -292,7 +291,7 @@ public class TezDagBuilder extends TezOp
payloadConf.set("pig.reduce.package", ObjectSerializer.serialize(pack));
payloadConf.set("pig.reduce.key.type", Byte.toString(keyType));
setIntermediateInputKeyValue(keyType, payloadConf);
- POPackage newPack;
+ POShuffleTezLoad newPack;
if (tezOp.isUnion()) {
newPack = new POUnionTezLoad(new OperatorKey(scope, nig.getNextNodeId(scope)), pack);
} else {
@@ -300,6 +299,13 @@ public class TezDagBuilder extends TezOp
}
tezOp.plan.add(newPack);
+ // Set input keys for POShuffleTezLoad. This is used to identify
+ // the inputs that are attached to the POShuffleTezLoad in the
+ // backend.
+ for (TezOperator pred : mPlan.getPredecessors(tezOp)) {
+ newPack.addInputKey(pred.getOperatorKey().toString());
+ }
+
if (succsList != null) {
for (PhysicalOperator succs : succsList) {
tezOp.plan.connect(newPack, succs);
Modified: pig/branches/tez/test/e2e/pig/tests/tez.conf
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/e2e/pig/tests/tez.conf?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/test/e2e/pig/tests/tez.conf (original)
+++ pig/branches/tez/test/e2e/pig/tests/tez.conf Wed Dec 18 18:51:11 2013
@@ -157,6 +157,26 @@ c = filter a by age < 20;
d = filter b by age < 20;
e = join c by name, d by name;
store e into ':OUTPATH:';\,
+ },
+ {
+ 'num' => 2,
+ 'pig' => q\set pig.tez.session.reuse false;
+a = load ':INPATH:/singlefile/studenttab20m' as (name, age, gpa);
+b = load ':INPATH:/singlefile/votertab10k' as (name, age, registration, contributions);
+c = filter a by age < 20;
+d = filter b by age < 20;
+e = join c by name, d by name using 'replicated';
+store e into ':OUTPATH:';\,
+ },
+ {
+ 'num' => 3,
+ 'pig' => q\set pig.tez.session.reuse false;
+a = load ':INPATH:/singlefile/studenttab20m' as (name, age, gpa);
+b = load ':INPATH:/singlefile/votertab10k' as (name, age, registration, contributions);
+c = filter a by age < 20;
+d = filter b by age < 20;
+e = join c by name LEFT OUTER, d by name using 'replicated';
+store e into ':OUTPATH:';\,
}
]
},
Added: pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC10.gld
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC10.gld?rev=1552058&view=auto
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC10.gld (added)
+++ pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC10.gld Wed Dec 18 18:51:11 2013
@@ -0,0 +1,67 @@
+#--------------------------------------------------
+# There are 1 DAGs in the session
+#--------------------------------------------------
+#--------------------------------------------------
+# TEZ DAG plan: scope-39
+#--------------------------------------------------
+Tez vertex scope-37 -> Tez vertex scope-36,
+Tez vertex scope-38 -> Tez vertex scope-36,
+Tez vertex scope-36
+
+Tez vertex scope-37
+# Plan on vertex
+Local Rearrange[tuple]{int}(false) - scope-30 -> scope-36
+| |
+| Project[int][0] - scope-25
+|
+|---b: New For Each(false,false)[bag] - scope-15
+ | |
+ | Cast[int] - scope-10
+ | |
+ | |---Project[bytearray][0] - scope-9
+ | |
+ | Cast[int] - scope-13
+ | |
+ | |---Project[bytearray][1] - scope-12
+ |
+ |---b: Load(file:///tmp/input2:org.apache.pig.builtin.PigStorage) - scope-8
+Tez vertex scope-38
+# Plan on vertex
+Local Rearrange[tuple]{int}(false) - scope-32 -> scope-36
+| |
+| Project[int][0] - scope-26
+|
+|---c: New For Each(false,false)[bag] - scope-23
+ | |
+ | Cast[int] - scope-18
+ | |
+ | |---Project[bytearray][0] - scope-17
+ | |
+ | Cast[int] - scope-21
+ | |
+ | |---Project[bytearray][1] - scope-20
+ |
+ |---c: Load(file:///tmp/input3:org.apache.pig.builtin.PigStorage) - scope-16
+Tez vertex scope-36
+# Plan on vertex
+d: Store(file:///tmp/output/d:org.apache.pig.builtin.PigStorage) - scope-35
+|
+|---d: FRJoin[tuple] - scope-27 <- scope-37,scope-38
+ | |
+ | Project[int][0] - scope-24
+ | |
+ | Project[int][0] - scope-25
+ | |
+ | Project[int][0] - scope-26
+ |
+ |---a: New For Each(false,false)[bag] - scope-7
+ | |
+ | Cast[int] - scope-2
+ | |
+ | |---Project[bytearray][0] - scope-1
+ | |
+ | Cast[int] - scope-5
+ | |
+ | |---Project[bytearray][1] - scope-4
+ |
+ |---a: Load(file:///tmp/input1:org.apache.pig.builtin.PigStorage) - scope-0
Added: pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC11.gld
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC11.gld?rev=1552058&view=auto
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC11.gld (added)
+++ pig/branches/tez/test/org/apache/pig/test/data/GoldenFiles/TEZC11.gld Wed Dec 18 18:51:11 2013
@@ -0,0 +1,89 @@
+#--------------------------------------------------
+# There are 1 DAGs in the session
+#--------------------------------------------------
+#--------------------------------------------------
+# TEZ DAG plan: scope-58
+#--------------------------------------------------
+Tez vertex scope-36 -> Tez vertex scope-37,
+Tez vertex scope-38 -> Tez vertex scope-37,
+Tez vertex scope-37
+
+Tez vertex scope-36
+# Plan on vertex
+b: Local Rearrange[tuple]{int}(false) - scope-51 -> scope-37
+| |
+| Project[int][0] - scope-53
+|
+|---b1: New For Each(false,false)[bag] - scope-39
+ | |
+ | Project[int][0] - scope-40
+ | |
+ | POUserFunc(org.apache.pig.builtin.COUNT$Initial)[tuple] - scope-41
+ | |
+ | |---Project[bag][1] - scope-42
+ | |
+ | |---Project[bag][1] - scope-43
+ |
+ |---Pre Combiner Local Rearrange[tuple]{Unknown} - scope-54
+ |
+ |---a: New For Each(false,false)[bag] - scope-7
+ | |
+ | Cast[int] - scope-2
+ | |
+ | |---Project[bytearray][0] - scope-1
+ | |
+ | Cast[int] - scope-5
+ | |
+ | |---Project[bytearray][1] - scope-4
+ |
+ |---a: Load(file:///tmp/input1:org.apache.pig.builtin.PigStorage) - scope-0
+Tez vertex scope-38
+# Plan on vertex
+Local Rearrange[tuple]{int}(false) - scope-32 -> scope-37
+| |
+| Project[int][0] - scope-28
+|
+|---c: New For Each(false,false)[bag] - scope-26
+ | |
+ | Cast[int] - scope-21
+ | |
+ | |---Project[bytearray][0] - scope-20
+ | |
+ | Cast[int] - scope-24
+ | |
+ | |---Project[bytearray][1] - scope-23
+ |
+ |---c: Load(file:///tmp/input2:org.apache.pig.builtin.PigStorage) - scope-19
+Tez vertex scope-37
+# Combine plan on edge <scope-36>
+b: Local Rearrange[tuple]{int}(false) - scope-55 -> scope-37
+| |
+| Project[int][0] - scope-57
+|
+|---b1: New For Each(false,false)[bag] - scope-44
+ | |
+ | Project[int][0] - scope-45
+ | |
+ | POUserFunc(org.apache.pig.builtin.COUNT$Intermediate)[tuple] - scope-46
+ | |
+ | |---Project[bag][1] - scope-47
+ |
+ |---b: Package(CombinerPackager)[tuple]{int} - scope-50
+# Plan on vertex
+d: Store(file:///tmp/output/e:org.apache.pig.builtin.PigStorage) - scope-35
+|
+|---d: FRJoin[tuple] - scope-29 <- scope-38
+ | |
+ | Project[int][0] - scope-27
+ | |
+ | Project[int][0] - scope-28
+ |
+ |---b1: New For Each(false,false)[bag] - scope-18
+ | |
+ | Project[int][0] - scope-12
+ | |
+ | POUserFunc(org.apache.pig.builtin.COUNT$Final)[long] - scope-16
+ | |
+ | |---Project[bag][1] - scope-48
+ |
+ |---b: Package(CombinerPackager)[tuple]{int} - scope-9
Modified: pig/branches/tez/test/org/apache/pig/tez/TestTezCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/tez/TestTezCompiler.java?rev=1552058&r1=1552057&r2=1552058&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/tez/TestTezCompiler.java (original)
+++ pig/branches/tez/test/org/apache/pig/tez/TestTezCompiler.java Wed Dec 18 18:51:11 2013
@@ -208,6 +208,33 @@ public class TestTezCompiler {
run(pp, "test/org/apache/pig/test/data/GoldenFiles/TEZC9.gld");
}
+ @Test
+ public void testReplicatedJoinInMapper() throws Exception {
+ String query =
+ "a = load 'file:///tmp/input1' as (x:int, y:int);" +
+ "b = load 'file:///tmp/input2' as (x:int, z:int);" +
+ "c = load 'file:///tmp/input3' as (x:int, z:int);" +
+ "d = join a by x, b by x, c by x using 'replicated';" +
+ "store d into 'file:///tmp/output/d';";
+
+ PhysicalPlan pp = Util.buildPp(pigServer, query);
+ run(pp, "test/org/apache/pig/test/data/GoldenFiles/TEZC10.gld");
+ }
+
+ @Test
+ public void testReplicatedJoinInReducer() throws Exception {
+ String query =
+ "a = load 'file:///tmp/input1' as (x:int, y:int);" +
+ "b = group a by x;" +
+ "b1 = foreach b generate group, COUNT(a.y);" +
+ "c = load 'file:///tmp/input2' as (x:int, z:int);" +
+ "d = join b1 by group, c by x using 'replicated';" +
+ "store d into 'file:///tmp/output/e';";
+
+ PhysicalPlan pp = Util.buildPp(pigServer, query);
+ run(pp, "test/org/apache/pig/test/data/GoldenFiles/TEZC11.gld");
+ }
+
private void run(PhysicalPlan pp, String expectedFile) throws Exception {
TezLauncher launcher = new TezLauncher();
TezPlanContainer tezPlanContainer = launcher.compile(pp, pc);