You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by zl...@apache.org on 2017/02/22 09:43:46 UTC
svn commit: r1783988 [11/24] - in /pig/branches/spark: ./ bin/ conf/
contrib/piggybank/java/
contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/
contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/util/apachel...
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POCounterStatsTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POCounterStatsTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POCounterStatsTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POCounterStatsTez.java Wed Feb 22 09:43:41 2017
@@ -56,6 +56,7 @@ public class POCounterStatsTez extends P
private transient KeyValuesReader reader;
private transient KeyValueWriter writer;
private transient boolean finished = false;
+ private transient boolean hasNext = false;
public POCounterStatsTez(OperatorKey k) {
super(k);
@@ -88,6 +89,7 @@ public class POCounterStatsTez extends P
try {
reader = (KeyValuesReader) input.getReader();
LOG.info("Attached input from vertex " + inputKey + " : input=" + input + ", reader=" + reader);
+ hasNext = reader.next();
} catch (Exception e) {
throw new ExecException(e);
}
@@ -130,12 +132,13 @@ public class POCounterStatsTez extends P
Integer key = null;
Long value = null;
// Read count of records per task
- while (reader.next()) {
+ while (hasNext) {
key = ((IntWritable)reader.getCurrentKey()).get();
for (Object val : reader.getCurrentValues()) {
value = ((LongWritable)val).get();
counterRecords.put(key, value);
}
+ hasNext = reader.next();
}
// BinInterSedes only takes String for map key
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POFRJoinTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POFRJoinTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POFRJoinTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POFRJoinTez.java Wed Feb 22 09:43:41 2017
@@ -19,6 +19,8 @@
package org.apache.pig.backend.hadoop.executionengine.tez.plan.operator;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -101,9 +103,13 @@ public class POFRJoinTez extends POFRJoi
LogicalInput input = inputs.get(key);
if (!this.replInputs.contains(input)) {
this.replInputs.add(input);
- this.replReaders.add((KeyValueReader) input.getReader());
+ KeyValueReader reader = (KeyValueReader) input.getReader();
+ this.replReaders.add(reader);
+ log.info("Attached input from vertex " + key + " : input=" + input + ", reader=" + reader);
}
}
+ // Do not force fetch input by reading first record. Cases like MultiQuery_Union_4 have
+ // multiple POFRJoinTez loading same replicate input and will skip records
} catch (Exception e) {
throw new ExecException(e);
}
@@ -114,6 +120,7 @@ public class POFRJoinTez extends POFRJoi
*
* @throws ExecException
*/
+ @SuppressWarnings("unchecked")
@Override
protected void setUpHashMap() throws ExecException {
@@ -121,8 +128,8 @@ public class POFRJoinTez extends POFRJoi
// where same POFRJoinTez occurs in different Split sub-plans
Object cacheValue = ObjectCache.getInstance().retrieve(cacheKey);
if (cacheValue != null) {
- replicates = (TupleToMapKey[]) cacheValue;
- log.info("Found " + (replicates.length - 1) + " replication hash tables in Tez cache. cachekey=" + cacheKey);
+ replicates = (List<Map<? extends Object, ? extends List<Tuple>>>) cacheValue;
+ log.info("Found " + (replicates.size() - 1) + " replication hash tables in Tez cache. cachekey=" + cacheKey);
return;
}
@@ -148,7 +155,7 @@ public class POFRJoinTez extends POFRJoi
long time1 = System.currentTimeMillis();
- replicates[fragment] = null;
+ replicates.set(fragment, null);
int inputIdx = 0;
// We need to adjust the index because the number of replInputs is
// one less than the number of inputSchemas. The inputSchemas
@@ -158,7 +165,12 @@ public class POFRJoinTez extends POFRJoi
SchemaTupleFactory inputSchemaTupleFactory = inputSchemaTupleFactories[schemaIdx];
SchemaTupleFactory keySchemaTupleFactory = keySchemaTupleFactories[schemaIdx];
- TupleToMapKey replicate = new TupleToMapKey(4000, keySchemaTupleFactory);
+ Map<Object, ArrayList<Tuple>> replicate;
+ if (keySchemaTupleFactory == null) {
+ replicate = new HashMap<Object, ArrayList<Tuple>>(4000);
+ } else {
+ replicate = new TupleToMapKey(4000, keySchemaTupleFactory);
+ }
POLocalRearrange lr = LRs[schemaIdx];
try {
@@ -168,7 +180,8 @@ public class POFRJoinTez extends POFRJoi
}
PigNullableWritable key = (PigNullableWritable) replReaders.get(inputIdx).getCurrentKey();
- if (isKeyNull(key.getValueAsPigType())) continue;
+ Object keyValue = key.getValueAsPigType();
+ if (isKeyNull(keyValue)) continue;
NullableTuple val = (NullableTuple) replReaders.get(inputIdx).getCurrentValue();
// POFRJoin#getValueTuple() is reused to construct valTuple,
@@ -176,27 +189,31 @@ public class POFRJoinTez extends POFRJoi
// construct one here.
Tuple retTuple = mTupleFactory.newTuple(3);
retTuple.set(0, key.getIndex());
- retTuple.set(1, key.getValueAsPigType());
+ retTuple.set(1, keyValue);
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));
+ ArrayList<Tuple> values = replicate.get(keyValue);
+ if (values == null) {
+ if (inputSchemaTupleFactory == null) {
+ values = new ArrayList<Tuple>(1);
+ } else {
+ values = new TuplesToSchemaTupleList(1, inputSchemaTupleFactory);
+ }
+ replicate.put(keyValue, values);
}
- replicate.get(keyTuple).add(valTuple);
+ values.add(valTuple);
}
} catch (IOException e) {
throw new ExecException(e);
}
- replicates[schemaIdx] = replicate;
+ replicates.set(schemaIdx, replicate);
inputIdx++;
schemaIdx++;
}
long time2 = System.currentTimeMillis();
- log.info((replicates.length - 1) + " replication hash tables built. Time taken: " + (time2 - time1));
+ log.info((replicates.size() - 1) + " replication hash tables built. Time taken: " + (time2 - time1));
ObjectCache.getInstance().cache(cacheKey, replicates);
log.info("Cached replicate hash tables in Tez ObjectRegistry with vertex scope. cachekey=" + cacheKey);
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POIdentityInOutTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POIdentityInOutTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POIdentityInOutTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POIdentityInOutTez.java Wed Feb 22 09:43:41 2017
@@ -57,6 +57,7 @@ public class POIdentityInOutTez extends
private transient KeyValuesReader shuffleReader;
private transient boolean shuffleInput;
private transient boolean finished = false;
+ private transient boolean hasNext = false;
public POIdentityInOutTez(OperatorKey k, POLocalRearrange inputRearrange, String inputKey) {
super(inputRearrange);
@@ -95,9 +96,12 @@ public class POIdentityInOutTez extends
Reader r = input.getReader();
if (r instanceof KeyValueReader) {
reader = (KeyValueReader) r;
+ // Force input fetch
+ hasNext = reader.next();
} else {
shuffleInput = true;
shuffleReader = (KeyValuesReader) r;
+ hasNext = shuffleReader.next();
}
LOG.info("Attached input from vertex " + inputKey + " : input=" + input + ", reader=" + r);
} catch (Exception e) {
@@ -127,7 +131,7 @@ public class POIdentityInOutTez extends
return RESULT_EOP;
}
if (shuffleInput) {
- while (shuffleReader.next()) {
+ while (hasNext) {
Object curKey = shuffleReader.getCurrentKey();
Iterable<Object> vals = shuffleReader.getCurrentValues();
if (isSkewedJoin) {
@@ -139,9 +143,10 @@ public class POIdentityInOutTez extends
for (Object val : vals) {
writer.write(curKey, val);
}
+ hasNext = shuffleReader.next();
}
} else {
- while (reader.next()) {
+ while (hasNext) {
if (isSkewedJoin) {
NullablePartitionWritable wrappedKey = new NullablePartitionWritable(
(PigNullableWritable) reader.getCurrentKey());
@@ -155,6 +160,7 @@ public class POIdentityInOutTez extends
writer.write(reader.getCurrentKey(),
reader.getCurrentValue());
}
+ hasNext = reader.next();
}
}
finished = true;
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POLocalRearrangeTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POLocalRearrangeTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POLocalRearrangeTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POLocalRearrangeTez.java Wed Feb 22 09:43:41 2017
@@ -71,8 +71,8 @@ public class POLocalRearrangeTez extends
}
}
- public String getOutputKey() {
- return outputKey;
+ public boolean containsOutputKey(String key) {
+ return outputKey.equals(key);
}
public void setOutputKey(String outputKey) {
@@ -122,6 +122,10 @@ public class POLocalRearrangeTez extends
}
}
+ protected Result getRearrangedTuple() throws ExecException {
+ return super.getNextTuple();
+ }
+
@Override
public Result getNextTuple() throws ExecException {
res = super.getNextTuple();
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/PORankTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/PORankTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/PORankTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/PORankTez.java Wed Feb 22 09:43:41 2017
@@ -51,6 +51,7 @@ public class PORankTez extends PORank im
private transient Map<Integer, Long> counterOffsets;
private transient Configuration conf;
private transient boolean finished = false;
+ private transient Boolean hasFirstRecord;
public PORankTez(PORank copy) {
super(copy);
@@ -100,6 +101,7 @@ public class PORankTez extends PORank im
try {
reader = (KeyValueReader) input.getReader();
LOG.info("Attached input from vertex " + tuplesInputKey + " : input=" + input + ", reader=" + reader);
+ hasFirstRecord = reader.next();
} catch (Exception e) {
throw new ExecException(e);
}
@@ -140,9 +142,18 @@ public class PORankTez extends PORank im
Result inp = null;
try {
- while (reader.next()) {
- inp = new Result(POStatus.STATUS_OK, reader.getCurrentValue());
- return addRank(inp);
+ if (hasFirstRecord != null) {
+ if (hasFirstRecord) {
+ hasFirstRecord = null;
+ inp = new Result(POStatus.STATUS_OK, reader.getCurrentValue());
+ return addRank(inp);
+ }
+ hasFirstRecord = null;
+ } else {
+ while (reader.next()) {
+ inp = new Result(POStatus.STATUS_OK, reader.getCurrentValue());
+ return addRank(inp);
+ }
}
} catch (IOException e) {
throw new ExecException(e);
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POShuffleTezLoad.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POShuffleTezLoad.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POShuffleTezLoad.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POShuffleTezLoad.java Wed Feb 22 09:43:41 2017
@@ -25,6 +25,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.WritableComparator;
import org.apache.pig.backend.executionengine.ExecException;
@@ -32,12 +34,16 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.AccumulativeTupleBuffer;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.CombinerPackager;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.LitePackager;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.Packager;
import org.apache.pig.backend.hadoop.executionengine.tez.runtime.TezInput;
import org.apache.pig.backend.hadoop.executionengine.util.AccumulatorOptimizerUtil;
import org.apache.pig.data.AccumulativeBag;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.InternalCachedBag;
+import org.apache.pig.data.ReadOnceBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.io.NullableTuple;
import org.apache.pig.impl.io.PigNullableWritable;
@@ -48,6 +54,7 @@ import org.apache.tez.runtime.library.co
public class POShuffleTezLoad extends POPackage implements TezInput {
private static final long serialVersionUID = 1L;
+ private static final Log LOG = LogFactory.getLog(POShuffleTezLoad.class);
protected List<String> inputKeys = new ArrayList<String>();
private boolean isSkewedJoin = false;
@@ -61,6 +68,7 @@ public class POShuffleTezLoad extends PO
private transient WritableComparator groupingComparator = null;
private transient Configuration conf;
private transient int accumulativeBatchSize;
+ private transient boolean readOnceOneBag;
public POShuffleTezLoad(POPackage pack) {
super(pack);
@@ -101,7 +109,10 @@ public class POShuffleTezLoad extends PO
// - Input key will be repeated, but index would be same within a TezInput
if (!this.inputs.contains(input)) {
this.inputs.add(input);
- this.readers.add((KeyValuesReader)input.getReader());
+ KeyValuesReader reader = (KeyValuesReader)input.getReader();
+ this.readers.add(reader);
+ LOG.info("Attached input from vertex " + inputKey
+ + " : input=" + input + ", reader=" + reader);
}
}
@@ -117,6 +128,13 @@ public class POShuffleTezLoad extends PO
for (int i = 0; i < numTezInputs; i++) {
finished[i] = !readers.get(i).next();
}
+
+ this.readOnceOneBag = (numInputs == 1)
+ && (pkgr instanceof CombinerPackager
+ || pkgr instanceof LitePackager || pkgr instanceof BloomPackager);
+ if (readOnceOneBag) {
+ readOnce[0] = true;
+ }
} catch (Exception e) {
throw new ExecException(e);
}
@@ -187,43 +205,47 @@ public class POShuffleTezLoad extends PO
} else {
- for (int i = 0; i < numInputs; i++) {
- bags[i] = new InternalCachedBag(numInputs);
- }
-
- if (numTezInputs == 1) {
- do {
- Iterable<Object> vals = readers.get(0).getCurrentValues();
- for (Object val : vals) {
- NullableTuple nTup = (NullableTuple) val;
- int index = nTup.getIndex();
- Tuple tup = pkgr.getValueTuple(keyWritable, nTup, index);
- bags[index].add(tup);
- }
- finished[0] = !readers.get(0).next();
- if (finished[0]) {
- break;
- }
- cur = readers.get(0).getCurrentKey();
- } while (groupingComparator.compare(min, cur) == 0); // We need to loop in case of Grouping Comparators
+ if (readOnceOneBag) {
+ bags[0] = new TezReadOnceBag(pkgr, min);
} else {
- for (int i = 0; i < numTezInputs; i++) {
- if (!finished[i]) {
- cur = readers.get(i).getCurrentKey();
- // We need to loop in case of Grouping Comparators
- while (groupingComparator.compare(min, cur) == 0) {
- Iterable<Object> vals = readers.get(i).getCurrentValues();
- for (Object val : vals) {
- NullableTuple nTup = (NullableTuple) val;
- int index = nTup.getIndex();
- Tuple tup = pkgr.getValueTuple(keyWritable, nTup, index);
- bags[index].add(tup);
- }
- finished[i] = !readers.get(i).next();
- if (finished[i]) {
- break;
- }
+ for (int i = 0; i < numInputs; i++) {
+ bags[i] = new InternalCachedBag(numInputs);
+ }
+
+ if (numTezInputs == 1) {
+ do {
+ Iterable<Object> vals = readers.get(0).getCurrentValues();
+ for (Object val : vals) {
+ NullableTuple nTup = (NullableTuple) val;
+ int index = nTup.getIndex();
+ Tuple tup = pkgr.getValueTuple(keyWritable, nTup, index);
+ bags[index].add(tup);
+ }
+ finished[0] = !readers.get(0).next();
+ if (finished[0]) {
+ break;
+ }
+ cur = readers.get(0).getCurrentKey();
+ } while (groupingComparator.compare(min, cur) == 0); // We need to loop in case of Grouping Comparators
+ } else {
+ for (int i = 0; i < numTezInputs; i++) {
+ if (!finished[i]) {
cur = readers.get(i).getCurrentKey();
+ // We need to loop in case of Grouping Comparators
+ while (groupingComparator.compare(min, cur) == 0) {
+ Iterable<Object> vals = readers.get(i).getCurrentValues();
+ for (Object val : vals) {
+ NullableTuple nTup = (NullableTuple) val;
+ int index = nTup.getIndex();
+ Tuple tup = pkgr.getValueTuple(keyWritable, nTup, index);
+ bags[index].add(tup);
+ }
+ finished[i] = !readers.get(i).next();
+ if (finished[i]) {
+ break;
+ }
+ cur = readers.get(i).getCurrentKey();
+ }
}
}
}
@@ -383,4 +405,74 @@ public class POShuffleTezLoad extends PO
}
+ private class TezReadOnceBag extends ReadOnceBag {
+
+ private static final long serialVersionUID = 1L;
+ private Iterator<Object> iter;
+
+ public TezReadOnceBag(Packager pkgr,
+ PigNullableWritable currentKey) throws IOException {
+ this.pkgr = pkgr;
+ this.keyWritable = currentKey;
+ this.iter = readers.get(0).getCurrentValues().iterator();
+ }
+
+ @Override
+ public Iterator<Tuple> iterator() {
+ return new TezReadOnceBagIterator();
+ }
+
+ private class TezReadOnceBagIterator implements Iterator<Tuple> {
+
+ @Override
+ public boolean hasNext() {
+ if (iter.hasNext()) {
+ return true;
+ } else {
+ try {
+ finished[0] = !readers.get(0).next();
+ if (finished[0]) {
+ return false;
+ }
+ // Currently combiner is not being applied when secondary key(grouping comparator) is used
+ // But might change in future. So check if the next key is same and return its values
+ Object cur = readers.get(0).getCurrentKey();
+ if (groupingComparator.compare(keyWritable, cur) == 0) {
+ iter = readers.get(0).getCurrentValues().iterator();
+ // Key should at least have one value. But doing a check just for safety
+ if (iter.hasNext()) {
+ return true;
+ } else {
+ throw new RuntimeException("Unexpected. Key " + keyWritable + " does not have any values");
+ }
+ }
+ return false;
+ } catch (IOException e) {
+ throw new RuntimeException("ReadOnceBag failed to get value tuple : ", e);
+ }
+ }
+ }
+
+ @Override
+ public Tuple next() {
+ NullableTuple ntup = (NullableTuple) iter.next();
+ int index = ntup.getIndex();
+ Tuple ret = null;
+ try {
+ ret = pkgr.getValueTuple(keyWritable, ntup, index);
+ } catch (ExecException e) {
+ throw new RuntimeException("ReadOnceBag failed to get value tuple : ", e);
+ }
+ return ret;
+ }
+
+ @Override
+ public void remove() {
+ throw new RuntimeException("ReadOnceBag.iterator().remove() is not allowed");
+ }
+ }
+
+ }
+
+
}
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POShuffledValueInputTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POShuffledValueInputTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POShuffledValueInputTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POShuffledValueInputTez.java Wed Feb 22 09:43:41 2017
@@ -57,6 +57,7 @@ public class POShuffledValueInputTez ext
private transient Iterator<KeyValueReader> readers;
private transient KeyValueReader currentReader;
private transient Configuration conf;
+ private transient Boolean hasFirstRecord;
public POShuffledValueInputTez(OperatorKey k) {
super(k);
@@ -98,6 +99,8 @@ public class POShuffledValueInputTez ext
}
readers = readersList.iterator();
currentReader = readers.next();
+ // Force input fetch
+ hasFirstRecord = currentReader.next();
} catch (Exception e) {
throw new ExecException(e);
}
@@ -111,7 +114,15 @@ public class POShuffledValueInputTez ext
}
do {
- if (currentReader.next()) {
+ if (hasFirstRecord != null) {
+ if (hasFirstRecord) {
+ hasFirstRecord = null;
+ Tuple origTuple = (Tuple) currentReader.getCurrentValue();
+ Tuple copy = mTupleFactory.newTuple(origTuple.getAll());
+ return new Result(POStatus.STATUS_OK, copy);
+ }
+ hasFirstRecord = null;
+ } else if (currentReader.next()) {
Tuple origTuple = (Tuple) currentReader.getCurrentValue();
Tuple copy = mTupleFactory.newTuple(origTuple.getAll());
return new Result(POStatus.STATUS_OK, copy);
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POSimpleTezLoad.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POSimpleTezLoad.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POSimpleTezLoad.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POSimpleTezLoad.java Wed Feb 22 09:43:41 2017
@@ -60,6 +60,8 @@ public class POSimpleTezLoad extends POL
private transient Configuration conf;
private transient boolean finished = false;
private transient TezCounter inputRecordCounter;
+ private transient boolean initialized;
+ private transient boolean noTupleCopy;
public POSimpleTezLoad(OperatorKey k, LoadFunc loader) {
super(k, loader);
@@ -149,7 +151,13 @@ public class POSimpleTezLoad extends POL
} else {
Result res = new Result();
Tuple next = (Tuple) reader.getCurrentValue();
- res.result = next;
+ if (!initialized) {
+ noTupleCopy = mTupleFactory.newTuple(1).getClass().isInstance(next);
+ initialized = true;
+ }
+ // Some Loaders return implementations of DefaultTuple instead of BinSedesTuple
+ // In that case copy to BinSedesTuple
+ res.result = noTupleCopy ? next : mTupleFactory.newTupleNoCopy(next.getAll());
res.returnStatus = POStatus.STATUS_OK;
if (inputRecordCounter != null) {
inputRecordCounter.increment(1);
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POStoreTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POStoreTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POStoreTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POStoreTez.java Wed Feb 22 09:43:41 2017
@@ -102,19 +102,19 @@ public class POStoreTez extends POStore
throw new ExecException(e);
}
- // Multiple outputs - can be another store or other outputs (shuffle, broadcast)
- if (outputs.size() > 1) {
- CounterGroup multiStoreGroup = processorContext.getCounters()
- .getGroup(MRPigStatsUtil.MULTI_STORE_COUNTER_GROUP);
- if (multiStoreGroup == null) {
- processorContext.getCounters().addGroup(
- MRPigStatsUtil.MULTI_STORE_COUNTER_GROUP,
- MRPigStatsUtil.MULTI_STORE_COUNTER_GROUP);
- }
- String name = MRPigStatsUtil.getMultiStoreCounterName(this);
- if (name != null) {
- outputRecordCounter = multiStoreGroup.addCounter(name, name, 0);
- }
+ // Even if there is a single hdfs output, we add multi store counter
+ // Makes it easier for user to see records for a particular store from
+ // the DAG counter
+ CounterGroup multiStoreGroup = processorContext.getCounters()
+ .getGroup(MRPigStatsUtil.MULTI_STORE_COUNTER_GROUP);
+ if (multiStoreGroup == null) {
+ processorContext.getCounters().addGroup(
+ MRPigStatsUtil.MULTI_STORE_COUNTER_GROUP,
+ MRPigStatsUtil.MULTI_STORE_COUNTER_GROUP);
+ }
+ String name = MRPigStatsUtil.getMultiStoreCounterName(this);
+ if (name != null) {
+ outputRecordCounter = multiStoreGroup.addCounter(name, name, 0);
}
}
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POValueInputTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POValueInputTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POValueInputTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/operator/POValueInputTez.java Wed Feb 22 09:43:41 2017
@@ -57,6 +57,7 @@ public class POValueInputTez extends Phy
private transient KeyValuesReader shuffleReader;
private transient boolean shuffleInput;
private transient boolean hasNext;
+ private transient Boolean hasFirstRecord;
public POValueInputTez(OperatorKey k) {
super(k);
@@ -92,6 +93,8 @@ public class POValueInputTez extends Phy
Reader r = input.getReader();
if (r instanceof KeyValueReader) {
reader = (KeyValueReader) r;
+ // Force input fetch
+ hasFirstRecord = reader.next();
} else {
shuffleInput = true;
shuffleReader = (KeyValuesReader) r;
@@ -118,10 +121,22 @@ public class POValueInputTez extends Phy
}
hasNext = shuffleReader.next();
}
- } else if (reader.next()) {
- Tuple origTuple = (Tuple) reader.getCurrentValue();
- Tuple copy = mTupleFactory.newTuple(origTuple.getAll());
- return new Result(POStatus.STATUS_OK, copy);
+ } else {
+ if (hasFirstRecord != null) {
+ if (hasFirstRecord) {
+ hasFirstRecord = null;
+ Tuple origTuple = (Tuple) reader.getCurrentValue();
+ Tuple copy = mTupleFactory.newTuple(origTuple.getAll());
+ return new Result(POStatus.STATUS_OK, copy);
+ }
+ hasFirstRecord = null;
+ } else {
+ while (reader.next()) {
+ Tuple origTuple = (Tuple) reader.getCurrentValue();
+ Tuple copy = mTupleFactory.newTuple(origTuple.getAll());
+ return new Result(POStatus.STATUS_OK, copy);
+ }
+ }
}
finished = true;
// For certain operators (such as STREAM), we could still have some work
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/CombinerOptimizer.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/CombinerOptimizer.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/CombinerOptimizer.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/CombinerOptimizer.java Wed Feb 22 09:43:41 2017
@@ -69,6 +69,11 @@ public class CombinerOptimizer extends T
}
for (TezOperator from : predecessors) {
+ PhysicalPlan combinePlan = to.inEdges.get(from.getOperatorKey()).combinePlan;
+ if (!combinePlan.isEmpty()) {
+ // Cases like bloom join have combine plan already set
+ continue;
+ }
List<POLocalRearrangeTez> rearranges = PlanHelper.getPhysicalOperators(from.plan, POLocalRearrangeTez.class);
if (rearranges.isEmpty()) {
continue;
@@ -77,7 +82,7 @@ public class CombinerOptimizer extends T
POLocalRearrangeTez connectingLR = null;
PhysicalPlan rearrangePlan = from.plan;
for (POLocalRearrangeTez lr : rearranges) {
- if (lr.getOutputKey().equals(to.getOperatorKey().toString())) {
+ if (lr.containsOutputKey(to.getOperatorKey().toString())) {
connectingLR = lr;
break;
}
@@ -90,7 +95,6 @@ public class CombinerOptimizer extends T
// Detected the POLocalRearrange -> POPackage pattern. Let's add
// combiner if possible.
- PhysicalPlan combinePlan = to.inEdges.get(from.getOperatorKey()).combinePlan;
CombinerOptimizerUtil.addCombiner(rearrangePlan, to.plan, combinePlan, messageCollector, doMapAgg);
if(!combinePlan.isEmpty()) {
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/LoaderProcessor.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/LoaderProcessor.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/LoaderProcessor.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/LoaderProcessor.java Wed Feb 22 09:43:41 2017
@@ -30,6 +30,7 @@ import org.apache.hadoop.mapreduce.Job;
import org.apache.pig.LoadFunc;
import org.apache.pig.PigConfiguration;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.InputSizeReducerEstimator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigInputFormat;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
@@ -65,11 +66,6 @@ public class LoaderProcessor extends Tez
this.jobConf.setBoolean("mapred.mapper.new-api", true);
this.jobConf.setClass("mapreduce.inputformat.class",
PigInputFormat.class, InputFormat.class);
- try {
- this.jobConf.set("pig.pigContext", ObjectSerializer.serialize(pc));
- } catch (IOException e) {
- throw new VisitorException(e);
- }
}
/**
@@ -175,6 +171,7 @@ public class LoaderProcessor extends Tez
// splits can be moved to if(loads) block below
int parallelism = tezOp.getLoaderInfo().getInputSplitInfo().getNumTasks();
tezOp.setRequestedParallelism(parallelism);
+ tezOp.setTotalInputFilesSize(InputSizeReducerEstimator.getTotalInputFileSize(conf, lds, job));
}
return lds;
}
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/MultiQueryOptimizerTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/MultiQueryOptimizerTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/MultiQueryOptimizerTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/MultiQueryOptimizerTez.java Wed Feb 22 09:43:41 2017
@@ -153,6 +153,8 @@ public class MultiQueryOptimizerTez exte
}
}
if (getPlan().getSuccessors(successor) != null) {
+ nonPackageInputSuccessors.clear();
+ toMergeSuccessors.clear();
for (TezOperator succSuccessor : getPlan().getSuccessors(successor)) {
if (succSuccessor.isUnion()) {
if (!(unionOptimizerOn &&
@@ -171,7 +173,13 @@ public class MultiQueryOptimizerTez exte
continue;
}
}
- toMergeSuccessors.add(succSuccessor);
+ if (TezCompilerUtil.isNonPackageInput(successor.getOperatorKey().toString(), succSuccessor)) {
+ // Output goes to scalar or POFRJoinTez in the union operator
+ // We need to ensure it is the only one to avoid parallel edges
+ canMerge = canMerge ? nonPackageInputSuccessors.add(succSuccessor) : false;
+ } else {
+ toMergeSuccessors.add(succSuccessor);
+ }
List<TezOperator> unionSuccessors = getPlan().getSuccessors(succSuccessor);
if (unionSuccessors != null) {
for (TezOperator unionSuccessor : unionSuccessors) {
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/ParallelismSetter.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/ParallelismSetter.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/ParallelismSetter.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/ParallelismSetter.java Wed Feb 22 09:43:41 2017
@@ -115,11 +115,16 @@ public class ParallelismSetter extends T
} else if (pc.defaultParallel != -1) {
parallelism = pc.defaultParallel;
}
+ if (parallelism == 0) {
+ // We need to produce empty output file.
+ // Even if user set PARALLEL 0, mapreduce has 1 reducer
+ parallelism = 1;
+ }
boolean overrideRequestedParallelism = false;
if (parallelism != -1
&& autoParallelismEnabled
- && tezOp.isIntermediateReducer()
&& !tezOp.isDontEstimateParallelism()
+ && tezOp.isIntermediateReducer()
&& tezOp.isOverrideIntermediateParallelism()) {
overrideRequestedParallelism = true;
}
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/SecondaryKeyOptimizerTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/SecondaryKeyOptimizerTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/SecondaryKeyOptimizerTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/SecondaryKeyOptimizerTez.java Wed Feb 22 09:43:41 2017
@@ -75,7 +75,7 @@ public class SecondaryKeyOptimizerTez ex
POLocalRearrangeTez connectingLR = null;
PhysicalPlan rearrangePlan = from.plan;
for (POLocalRearrangeTez lr : rearranges) {
- if (lr.getOutputKey().equals(to.getOperatorKey().toString())) {
+ if (lr.containsOutputKey(to.getOperatorKey().toString())) {
connectingLR = lr;
break;
}
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/TezEstimatedParallelismClearer.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/TezEstimatedParallelismClearer.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/TezEstimatedParallelismClearer.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/TezEstimatedParallelismClearer.java Wed Feb 22 09:43:41 2017
@@ -30,6 +30,8 @@ public class TezEstimatedParallelismClea
@Override
public void visitTezOp(TezOperator tezOp) throws VisitorException {
- tezOp.setEstimatedParallelism(-1);
+ if (!tezOp.isDontEstimateParallelism()) {
+ tezOp.setEstimatedParallelism(-1);
+ }
}
}
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/TezOperDependencyParallelismEstimator.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/TezOperDependencyParallelismEstimator.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/TezOperDependencyParallelismEstimator.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/TezOperDependencyParallelismEstimator.java Wed Feb 22 09:43:41 2017
@@ -62,7 +62,6 @@ import org.apache.tez.dag.api.EdgeProper
*/
public class TezOperDependencyParallelismEstimator implements TezParallelismEstimator {
- static private int maxTaskCount;
static final double DEFAULT_FLATTEN_FACTOR = 10;
static final double DEFAULT_FILTER_FACTOR = 0.7;
static final double DEFAULT_LIMIT_FACTOR = 0.1;
@@ -76,6 +75,8 @@ public class TezOperDependencyParallelis
static final double DEFAULT_AGGREGATION_FACTOR = 0.7;
private PigContext pc;
+ private int maxTaskCount;
+ private long bytesPerReducer;
@Override
public void setPigContext(PigContext pc) {
@@ -94,16 +95,18 @@ public class TezOperDependencyParallelis
maxTaskCount = conf.getInt(PigReducerEstimator.MAX_REDUCER_COUNT_PARAM,
PigReducerEstimator.DEFAULT_MAX_REDUCER_COUNT_PARAM);
- // If parallelism is set explicitly, respect it
- if (!tezOper.isIntermediateReducer() && tezOper.getRequestedParallelism()!=-1) {
- return tezOper.getRequestedParallelism();
- }
+ bytesPerReducer = conf.getLong(PigReducerEstimator.BYTES_PER_REDUCER_PARAM, PigReducerEstimator.DEFAULT_BYTES_PER_REDUCER);
// If we have already estimated parallelism, use that one
- if (tezOper.getEstimatedParallelism()!=-1) {
+ if (tezOper.getEstimatedParallelism() != -1) {
return tezOper.getEstimatedParallelism();
}
+ // If parallelism is set explicitly, respect it
+ if (!tezOper.isIntermediateReducer() && tezOper.getRequestedParallelism()!=-1) {
+ return tezOper.getRequestedParallelism();
+ }
+
List<TezOperator> preds = plan.getPredecessors(tezOper);
if (preds==null) {
throw new IOException("Cannot estimate parallelism for source vertex");
@@ -130,6 +133,12 @@ public class TezOperDependencyParallelis
boolean applyFactor = !tezOper.isUnion();
if (!pred.isVertexGroup() && applyFactor) {
predParallelism = predParallelism * pred.getParallelismFactor(tezOper);
+ if (pred.getTotalInputFilesSize() > 0) {
+ // Estimate similar to mapreduce and use the maximum of two
+ int parallelismBySize = (int) Math.ceil((double) pred
+ .getTotalInputFilesSize() / bytesPerReducer);
+ predParallelism = Math.max(predParallelism, parallelismBySize);
+ }
}
estimatedParallelism += predParallelism;
}
@@ -157,9 +166,7 @@ public class TezOperDependencyParallelis
}
if (roundedEstimatedParallelism == 0) {
- throw new IOException("Estimated parallelism for "
- + tezOper.getOperatorKey().toString()
- + " is 0 which is unexpected");
+ roundedEstimatedParallelism = 1; // We need to produce empty output file
}
return roundedEstimatedParallelism;
@@ -196,7 +203,7 @@ public class TezOperDependencyParallelis
if (successor != null) {
// Map side combiner
TezEdgeDescriptor edge = tezOp.outEdges.get(successor.getOperatorKey());
- if (!edge.combinePlan.isEmpty()) {
+ if (!edge.combinePlan.isEmpty() || edge.needsDistinctCombiner()) {
if (successor.isDistinct()) {
factor = DEFAULT_DISTINCT_FACTOR;
} else {
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/UnionOptimizer.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/UnionOptimizer.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/UnionOptimizer.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/UnionOptimizer.java Wed Feb 22 09:43:41 2017
@@ -29,6 +29,7 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pig.PigConfiguration;
+import org.apache.pig.StoreFunc;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
@@ -44,6 +45,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.tez.plan.operator.POStoreTez;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.operator.POValueOutputTez;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.udf.ReadScalarsTez;
+import org.apache.pig.backend.hadoop.executionengine.tez.runtime.HashValuePartitioner;
import org.apache.pig.backend.hadoop.executionengine.tez.runtime.TezInput;
import org.apache.pig.backend.hadoop.executionengine.tez.runtime.TezOutput;
import org.apache.pig.backend.hadoop.executionengine.tez.util.TezCompilerUtil;
@@ -52,7 +54,6 @@ import org.apache.pig.builtin.AvroStorag
import org.apache.pig.builtin.JsonStorage;
import org.apache.pig.builtin.OrcStorage;
import org.apache.pig.builtin.PigStorage;
-import org.apache.pig.builtin.RoundRobinPartitioner;
import org.apache.pig.builtin.mock.Storage;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.impl.plan.PlanException;
@@ -108,6 +109,12 @@ public class UnionOptimizer extends TezO
if((tezOp.isLimit() || tezOp.isLimitAfterSort()) && tezOp.getRequestedParallelism() == 1) {
return false;
}
+
+ // If user has specified a PARALLEL clause with the union operator
+ // turn off union optimization
+ if (tezOp.getRequestedParallelism() != -1) {
+ return false;
+ }
// Two vertices separately ranking with 1 to n and writing to output directly
// will make each rank repeate twice which is wrong. Rank always needs to be
// done from single vertex to have the counting correct.
@@ -120,10 +127,25 @@ public class UnionOptimizer extends TezO
public static boolean isOptimizableStoreFunc(TezOperator tezOp,
List<String> supportedStoreFuncs, List<String> unsupportedStoreFuncs)
throws VisitorException {
- if (supportedStoreFuncs != null || unsupportedStoreFuncs != null) {
- List<POStoreTez> stores = PlanHelper.getPhysicalOperators(tezOp.plan, POStoreTez.class);
- for (POStoreTez store : stores) {
- String name = store.getStoreFunc().getClass().getName();
+ List<POStoreTez> stores = PlanHelper.getPhysicalOperators(tezOp.plan, POStoreTez.class);
+
+ for (POStoreTez store : stores) {
+ String name = store.getStoreFunc().getClass().getName();
+ if (store.getStoreFunc() instanceof StoreFunc) {
+ StoreFunc func = (StoreFunc) store.getStoreFunc();
+ if (func.supportsParallelWriteToStoreLocation() != null) {
+ if (func.supportsParallelWriteToStoreLocation()) {
+ continue;
+ } else {
+ LOG.warn(name + " does not support union optimization."
+ + " Disabling it. There will be some performance degradation.");
+ return false;
+ }
+ }
+ }
+ // If StoreFunc does not explicitly state support, then check supported and
+ // unsupported config settings.
+ if (supportedStoreFuncs != null || unsupportedStoreFuncs != null) {
if (unsupportedStoreFuncs != null
&& unsupportedStoreFuncs.contains(name)) {
return false;
@@ -237,8 +259,23 @@ public class UnionOptimizer extends TezO
for (TezOperator succ : successors) {
if (succ.isVertexGroup() && unionStoreOutputs.get(i).getSFile().equals(succ.getVertexGroupInfo().getSFile())) {
existingVertexGroup = succ;
+ break;
+ }
+ }
+ }
+ if (existingVertexGroup == null) {
+ // In the case of union + split + union + store, the different stores in the Split
+ // will be writing to same location after second union operator is optimized.
+ // So while optimizing the first union, we should just make it write to one vertex group
+ for (int j = 0; j < i; j++) {
+ if (unionStoreOutputs.get(i).getSFile().equals(storeVertexGroupOps[j].getVertexGroupInfo().getSFile())) {
+ storeVertexGroupOps[i] = storeVertexGroupOps[j];
+ break;
}
}
+ if (storeVertexGroupOps[i] != null) {
+ continue;
+ }
}
if (existingVertexGroup != null) {
storeVertexGroupOps[i] = existingVertexGroup;
@@ -270,6 +307,15 @@ public class UnionOptimizer extends TezO
TezOperator[] outputVertexGroupOps = new TezOperator[unionOutputKeys.size()];
String[] newOutputKeys = new String[unionOutputKeys.size()];
for (int i=0; i < outputVertexGroupOps.length; i++) {
+ for (int j = 0; j < i; j++) {
+ if (unionOutputKeys.get(i).equals(unionOutputKeys.get(j))) {
+ outputVertexGroupOps[i] = outputVertexGroupOps[j];
+ break;
+ }
+ }
+ if (outputVertexGroupOps[i] != null) {
+ continue;
+ }
outputVertexGroupOps[i] = new TezOperator(OperatorKey.genOpKey(scope));
outputVertexGroupOps[i].setVertexGroupInfo(new VertexGroupInfo());
outputVertexGroupOps[i].getVertexGroupInfo().setOutput(unionOutputKeys.get(i));
@@ -515,15 +561,24 @@ public class UnionOptimizer extends TezO
// Connect predecessor to the storeVertexGroups
int i = 0;
for (TezOperator storeVertexGroup : storeVertexGroupOps) {
+ // Skip connecting if they are already connected. Can happen in case of
+ // union + split + union + store. Because of the split all the stores
+ // will be writing to same location
+ List<OperatorKey> inputs = storeVertexGroup.getVertexGroupInfo().getInputs();
+ if (inputs == null || !inputs.contains(pred.getOperatorKey())) {
+ tezPlan.connect(pred, storeVertexGroup);
+ }
storeVertexGroup.getVertexGroupInfo().addInput(pred.getOperatorKey());
pred.addVertexGroupStore(clonedUnionStoreOutputs.get(i++).getOperatorKey(),
storeVertexGroup.getOperatorKey());
- tezPlan.connect(pred, storeVertexGroup);
}
for (TezOperator outputVertexGroup : outputVertexGroupOps) {
+ List<OperatorKey> inputs = outputVertexGroup.getVertexGroupInfo().getInputs();
+ if (inputs == null || !inputs.contains(pred.getOperatorKey())) {
+ tezPlan.connect(pred, outputVertexGroup);
+ }
outputVertexGroup.getVertexGroupInfo().addInput(pred.getOperatorKey());
- tezPlan.connect(pred, outputVertexGroup);
}
copyOperatorProperties(pred, unionOp);
@@ -568,7 +623,7 @@ public class UnionOptimizer extends TezO
// more union predecessors. Change it to SCATTER_GATHER
if (edge.dataMovementType == DataMovementType.ONE_TO_ONE) {
edge.dataMovementType = DataMovementType.SCATTER_GATHER;
- edge.partitionerClass = RoundRobinPartitioner.class;
+ edge.partitionerClass = HashValuePartitioner.class;
edge.outputClassName = UnorderedPartitionedKVOutput.class.getName();
edge.inputClassName = UnorderedKVInput.class.getName();
}
Added: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/HashValuePartitioner.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/HashValuePartitioner.java?rev=1783988&view=auto
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/HashValuePartitioner.java (added)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/HashValuePartitioner.java Wed Feb 22 09:43:41 2017
@@ -0,0 +1,58 @@
+/**
+ * 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.runtime;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.Partitioner;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.io.NullableTuple;
+
+public class HashValuePartitioner extends Partitioner<Writable, Writable> {
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ public int getPartition(Writable key, Writable value, int numPartitions) {
+ int hash = 17;
+ Tuple tuple;
+ if (value instanceof Tuple) {
+ // union optimizer turned off
+ tuple = (Tuple) value;
+ } else {
+ // union followed by order by or skewed join
+ tuple = (Tuple)((NullableTuple) value).getValueAsPigType();
+ }
+ if (tuple != null) {
+ for (Object o : tuple.getAll()) {
+ if (o != null) {
+ // Skip computing hashcode for bags.
+ // Order of elements in the map/bag may be different on each run
+ // Can't even include size as some DataBag implementations
+ // iterate through all elements in the bag to get the size.
+ if (o instanceof DataBag) {
+ hash = 31 * hash;
+ } else {
+ hash = 31 * hash + o.hashCode();
+ }
+ }
+ }
+ }
+ return (hash & Integer.MAX_VALUE) % numPartitions;
+ }
+
+}
\ No newline at end of file
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PartitionerDefinedVertexManager.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PartitionerDefinedVertexManager.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PartitionerDefinedVertexManager.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PartitionerDefinedVertexManager.java Wed Feb 22 09:43:41 2017
@@ -17,23 +17,25 @@
*/
package org.apache.pig.backend.hadoop.executionengine.tez.runtime;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.tez.dag.api.EdgeManagerPluginDescriptor;
import org.apache.tez.dag.api.EdgeProperty;
import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.dag.api.InputDescriptor;
import org.apache.tez.dag.api.VertexManagerPlugin;
import org.apache.tez.dag.api.VertexManagerPluginContext;
import org.apache.tez.dag.api.VertexManagerPluginContext.TaskWithLocationHint;
-import org.apache.tez.dag.app.dag.impl.ScatterGatherEdgeManager;
+import org.apache.tez.dag.api.event.VertexState;
+import org.apache.tez.dag.api.event.VertexStateUpdate;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.events.VertexManagerEvent;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
/**
@@ -46,8 +48,13 @@ import com.google.common.collect.Lists;
public class PartitionerDefinedVertexManager extends VertexManagerPlugin {
private static final Log LOG = LogFactory.getLog(PartitionerDefinedVertexManager.class);
- private boolean isParallelismSet = false;
+ private volatile boolean parallelismSet;
private int dynamicParallelism = -1;
+ private int numConfiguredSources;
+ private int numSources = -1;
+ private volatile boolean configured;
+ private volatile boolean started;
+ private volatile boolean scheduled;
public PartitionerDefinedVertexManager(VertexManagerPluginContext context) {
super(context);
@@ -55,7 +62,31 @@ public class PartitionerDefinedVertexMan
@Override
public void initialize() {
- // Nothing to do
+ // this will prevent vertex from starting until we notify we are done
+ getContext().vertexReconfigurationPlanned();
+ parallelismSet = false;
+ numConfiguredSources = 0;
+ configured = false;
+ started = false;
+ numSources = getContext().getInputVertexEdgeProperties().size();
+ // wait for sources and self to start
+ Map<String, EdgeProperty> edges = getContext().getInputVertexEdgeProperties();
+ for (String entry : edges.keySet()) {
+ getContext().registerForVertexStateUpdates(entry, EnumSet.of(VertexState.CONFIGURED));
+ }
+ }
+
+ @Override
+ public synchronized void onVertexStateUpdated(VertexStateUpdate stateUpdate)
+ throws Exception {
+ numConfiguredSources++;
+ LOG.info("For vertex: " + getContext().getVertexName() + " Received configured signal from: "
+ + stateUpdate.getVertexName() + " numConfiguredSources: " + numConfiguredSources
+ + " needed: " + numSources);
+ Preconditions.checkState(numConfiguredSources <= numSources, "Vertex: " + getContext().getVertexName());
+ if (numConfiguredSources == numSources) {
+ configure();
+ }
}
@Override
@@ -73,10 +104,9 @@ public class PartitionerDefinedVertexMan
public void onVertexManagerEventReceived(VertexManagerEvent vmEvent) throws Exception {
// There could be multiple partition vertex sending VertexManagerEvent
// Only need to setVertexParallelism once
- if (isParallelismSet) {
+ if (parallelismSet) {
return;
}
- isParallelismSet = true;
// Need to distinguish from VertexManagerEventPayloadProto emitted by OrderedPartitionedKVOutput
if (vmEvent.getUserPayload().limit()==4) {
dynamicParallelism = vmEvent.getUserPayload().getInt();
@@ -96,18 +126,50 @@ public class PartitionerDefinedVertexMan
edgeManagers.put(entry.getKey(), edge);
}
getContext().reconfigureVertex(dynamicParallelism, null, edgeManagers);
+ parallelismSet = true;
+ configure();
}
}
}
- @Override
- public void onVertexStarted(Map<String, List<Integer>> completions) {
- if (dynamicParallelism != -1) {
+ private void configure() {
+ if(parallelismSet && (numSources == numConfiguredSources)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Done reconfiguring vertex " + getContext().getVertexName());
+ }
+ getContext().doneReconfiguringVertex();
+ configured = true;
+ trySchedulingTasks();
+ }
+ }
+
+ private synchronized void trySchedulingTasks() {
+ if (configured && started && !scheduled) {
+ LOG.info("Scheduling " + dynamicParallelism + " tasks for vertex " + getContext().getVertexName());
List<TaskWithLocationHint> tasksToStart = Lists.newArrayListWithCapacity(dynamicParallelism);
- for (int i=0; i<dynamicParallelism; ++i) {
+ for (int i = 0; i < dynamicParallelism; ++i) {
tasksToStart.add(new TaskWithLocationHint(new Integer(i), null));
}
getContext().scheduleVertexTasks(tasksToStart);
+ scheduled = true;
}
}
+
+ @Override
+ public void onVertexStarted(Map<String, List<Integer>> completions) {
+ // This vertex manager will be getting the following calls
+ // 1) onVertexManagerEventReceived - Parallelism vertex manager event sent by sample aggregator vertex
+ // 2) onVertexStateUpdated - Vertex CONFIGURED status updates from
+ // - Order by Partitioner vertex (1-1) in case of Order by
+ // - Skewed Join Left Partitioner (1-1) and Right Input Vertices in case of SkewedJoin
+ // 3) onVertexStarted
+ // Calls 2) and 3) can happen in any order. So we should schedule tasks
+ // only after start is called and configuration is also complete
+ started = true;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Vertex start received for " + getContext().getVertexName());
+ }
+ trySchedulingTasks();
+ }
+
}
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java Wed Feb 22 09:43:41 2017
@@ -33,15 +33,15 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.tez.plan.optimizer.ParallelismSetter;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.optimizer.TezEstimatedParallelismClearer;
import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.PigImplConstants;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.ObjectSerializer;
import org.apache.tez.common.TezUtils;
import org.apache.tez.dag.api.EdgeProperty;
-import org.apache.tez.dag.api.TezException;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.dag.api.VertexManagerPluginContext;
-import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.dag.api.event.VertexState;
import org.apache.tez.dag.api.event.VertexStateUpdate;
import org.apache.tez.dag.library.vertexmanager.ShuffleVertexManager;
@@ -72,7 +72,7 @@ public class PigGraceShuffleVertexManage
conf = TezUtils.createConfFromUserPayload(getContext().getUserPayload());
bytesPerTask = conf.getLong(InputSizeReducerEstimator.BYTES_PER_REDUCER_PARAM,
InputSizeReducerEstimator.DEFAULT_BYTES_PER_REDUCER);
- pc = (PigContext)ObjectSerializer.deserialize(conf.get("pig.pigContext"));
+ pc = (PigContext)ObjectSerializer.deserialize(conf.get(PigImplConstants.PIG_CONTEXT));
tezPlan = (TezOperPlan)ObjectSerializer.deserialize(conf.get("pig.tez.plan"));
TezEstimatedParallelismClearer clearer = new TezEstimatedParallelismClearer(tezPlan);
try {
@@ -81,9 +81,10 @@ public class PigGraceShuffleVertexManage
throw new TezUncheckedException(e);
}
TezOperator op = tezPlan.getOperator(OperatorKey.fromString(getContext().getVertexName()));
-
+
// Collect grandparents of the vertex
- Function<TezOperator, String> tezOpToString = new Function<TezOperator, String>() {
+ Function<TezOperator, String> tezOpToString = new Function<TezOperator, String>() {
+ @Override
public String apply(TezOperator op) { return op.getOperatorKey().toString(); }
};
grandParents = Lists.transform(TezOperPlan.getGrandParentsForGraceParallelism(tezPlan, op), tezOpToString);
@@ -135,7 +136,7 @@ public class PigGraceShuffleVertexManage
// Now one of the predecessor is about to start, we need to make a decision now
if (anyPredAboutToStart) {
// All grandparents finished, start parents with right parallelism
-
+
for (TezOperator pred : preds) {
if (pred.getRequestedParallelism()==-1) {
List<TezOperator> predPreds = tezPlan.getPredecessors(pred);
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java Wed Feb 22 09:43:41 2017
@@ -25,6 +25,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Properties;
import java.util.Set;
import org.apache.commons.logging.Log;
@@ -32,6 +33,7 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.TaskAttemptID;
import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.log4j.PropertyConfigurator;
import org.apache.pig.JVMReuseImpl;
import org.apache.pig.PigConstants;
import org.apache.pig.PigException;
@@ -39,6 +41,7 @@ import org.apache.pig.backend.executione
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigHadoopLogger;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigInputFormat;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.ProgressableReporter;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.UDFFinishVisitor;
@@ -53,6 +56,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.data.SchemaTupleBackend;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.PigImplConstants;
import org.apache.pig.impl.plan.DependencyOrderWalker;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.ObjectSerializer;
@@ -132,7 +136,11 @@ public class PigProcessor extends Abstra
SpillableMemoryManager.getInstance().configure(conf);
PigContext.setPackageImportList((ArrayList<String>) ObjectSerializer
.deserialize(conf.get("udf.import.list")));
- PigContext pc = (PigContext) ObjectSerializer.deserialize(conf.get("pig.pigContext"));
+ Properties log4jProperties = (Properties) ObjectSerializer
+ .deserialize(conf.get(PigImplConstants.PIG_LOG4J_PROPERTIES));
+ if (log4jProperties != null) {
+ PropertyConfigurator.configure(log4jProperties);
+ }
// To determine front-end in UDFContext
conf.set(MRConfiguration.JOB_APPLICATION_ATTEMPT_ID, getContext().getUniqueIdentifier());
@@ -151,6 +159,12 @@ public class PigProcessor extends Abstra
conf.setInt(JobContext.TASK_PARTITION,
taskAttemptId.getTaskID().getId());
conf.set(JobContext.ID, taskAttemptId.getJobID().toString());
+ if (conf.get(PigInputFormat.PIG_INPUT_LIMITS) != null) {
+ // Has Load and is a root vertex
+ conf.setInt(JobContext.NUM_MAPS, getContext().getVertexParallelism());
+ } else {
+ conf.setInt(JobContext.NUM_REDUCES, getContext().getVertexParallelism());
+ }
conf.set(PigConstants.TASK_INDEX, Integer.toString(getContext().getTaskIndex()));
UDFContext.getUDFContext().addJobConf(conf);
@@ -158,7 +172,7 @@ public class PigProcessor extends Abstra
String execPlanString = conf.get(PLAN);
execPlan = (PhysicalPlan) ObjectSerializer.deserialize(execPlanString);
- SchemaTupleBackend.initialize(conf, pc);
+ SchemaTupleBackend.initialize(conf);
PigMapReduce.sJobContext = HadoopShims.createJobContext(conf, new org.apache.hadoop.mapreduce.JobID());
// Set the job conf as a thread-local member of PigMapReduce
@@ -167,7 +181,7 @@ public class PigProcessor extends Abstra
Utils.setDefaultTimeZone(conf);
- boolean aggregateWarning = "true".equalsIgnoreCase(pc.getProperties().getProperty("aggregate.warning"));
+ boolean aggregateWarning = "true".equalsIgnoreCase(conf.get("aggregate.warning"));
PigStatusReporter pigStatusReporter = PigStatusReporter.getInstance();
pigStatusReporter.setContext(new TezTaskContext(getContext()));
pigHadoopLogger = PigHadoopLogger.getInstance();
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/TezInput.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/TezInput.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/TezInput.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/TezInput.java Wed Feb 22 09:43:41 2017
@@ -43,6 +43,15 @@ public interface TezInput {
*/
public void addInputsToSkip(Set<String> inputsToSkip);
+ /**
+ * Attach the inputs to the operator. Also ensure reader.next() is called to force fetch
+ * the input so that all inputs are fetched and memory released before memory is allocated
+ * for outputs
+ *
+ * @param inputs available inputs
+ * @param conf configuration
+ * @throws ExecException
+ */
public void attachInputs(Map<String, LogicalInput> inputs,
Configuration conf) throws ExecException;
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/WeightedRangePartitionerTez.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/WeightedRangePartitionerTez.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/WeightedRangePartitionerTez.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/WeightedRangePartitionerTez.java Wed Feb 22 09:43:41 2017
@@ -23,6 +23,7 @@ import java.util.Map.Entry;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.JobContext;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.partitioners.DiscreteProbabilitySampleGenerator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.partitioners.WeightedRangePartitioner;
import org.apache.pig.data.DataBag;
@@ -30,6 +31,7 @@ import org.apache.pig.data.InternalMap;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.builtin.FindQuantiles;
import org.apache.pig.impl.io.PigNullableWritable;
+import org.apache.pig.impl.util.UDFContext;
import org.apache.tez.runtime.library.common.ConfigUtils;
public class WeightedRangePartitionerTez extends WeightedRangePartitioner {
@@ -64,11 +66,13 @@ public class WeightedRangePartitionerTez
InternalMap weightedPartsData = (InternalMap) quantileMap.get(FindQuantiles.WEIGHTED_PARTS);
estimatedNumPartitions = (Integer)quantileMap.get(PigProcessor.ESTIMATED_NUM_PARALLELISM);
convertToArray(quantilesList);
+ long taskIdHashCode = UDFContext.getUDFContext().getJobConf().get(JobContext.TASK_ID).hashCode();
+ long randomSeed = ((long)taskIdHashCode << 32) | (taskIdHashCode & 0xffffffffL);
for (Entry<Object, Object> ent : weightedPartsData.entrySet()) {
Tuple key = (Tuple) ent.getKey(); // sample item which repeats
float[] probVec = getProbVec((Tuple) ent.getValue());
weightedParts.put(getPigNullableWritable(key),
- new DiscreteProbabilitySampleGenerator(probVec));
+ new DiscreteProbabilitySampleGenerator(randomSeed, probVec));
}
} catch (Exception e) {
throw new RuntimeException(e);
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java Wed Feb 22 09:43:41 2017
@@ -50,6 +50,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.classification.InterfaceAudience;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.library.vertexmanager.ShuffleVertexManager;
import org.apache.tez.mapreduce.hadoop.DeprecatedKeys;
import org.apache.tez.mapreduce.hadoop.InputSplitInfo;
import org.apache.tez.mapreduce.hadoop.InputSplitInfoDisk;
@@ -102,7 +103,6 @@ public class MRToTezHelper {
mrReduceParamToTezVertexParamMap.put(MRJobConfig.REDUCE_SPECULATIVE, TezConfiguration.TEZ_AM_SPECULATION_ENABLED);
mrReduceParamToTezVertexParamMap.put(MRJobConfig.REDUCE_LOG_LEVEL, TezConfiguration.TEZ_TASK_LOG_LEVEL);
mrReduceParamToTezVertexParamMap.put("mapreduce.job.running.reduce.limit", "tez.am.vertex.max-task-concurrency");
- mrReduceParamToTezVertexParamMap.put("mapreduce.job.running.map.limit", "tez.am.vertex.max-task-concurrency");
mrReduceParamToTezVertexParamMap.put(MRJobConfig.TASK_TIMEOUT, "tez.am.progress.stuck.interval-ms");
}
@@ -165,11 +165,7 @@ public class MRToTezHelper {
continue;
}
}
- if (key.startsWith("dfs.datanode")) {
- tezConf.unset(key);
- } else if (key.startsWith("dfs.namenode")) {
- tezConf.unset(key);
- } else if (key.startsWith("yarn.nodemanager")) {
+ if (key.startsWith("yarn.nodemanager")) {
tezConf.unset(key);
} else if (key.startsWith("mapreduce.jobhistory")) {
tezConf.unset(key);
@@ -181,20 +177,15 @@ public class MRToTezHelper {
}
}
- public static TezConfiguration getDAGAMConfFromMRConf(
- Configuration tezConf) {
-
- // Set Tez parameters based on MR parameters.
- TezConfiguration dagAMConf = new TezConfiguration(tezConf);
-
+ public static void translateMRSettingsForTezAM(TezConfiguration dagAMConf) {
convertMRToTezConf(dagAMConf, dagAMConf, DeprecatedKeys.getMRToDAGParamMap());
convertMRToTezConf(dagAMConf, dagAMConf, mrAMParamToTezAMParamMap);
- String env = tezConf.get(MRJobConfig.MR_AM_ADMIN_USER_ENV);
- if (tezConf.get(MRJobConfig.MR_AM_ENV) != null) {
- env = (env == null) ? tezConf.get(MRJobConfig.MR_AM_ENV)
- : env + "," + tezConf.get(MRJobConfig.MR_AM_ENV);
+ String env = dagAMConf.get(MRJobConfig.MR_AM_ADMIN_USER_ENV);
+ if (dagAMConf.get(MRJobConfig.MR_AM_ENV) != null) {
+ env = (env == null) ? dagAMConf.get(MRJobConfig.MR_AM_ENV)
+ : env + "," + dagAMConf.get(MRJobConfig.MR_AM_ENV);
}
if (env != null) {
@@ -203,24 +194,23 @@ public class MRToTezHelper {
dagAMConf.setIfUnset(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS,
org.apache.tez.mapreduce.hadoop.MRHelpers
- .getJavaOptsForMRAM(tezConf));
+ .getJavaOptsForMRAM(dagAMConf));
- String queueName = tezConf.get(JobContext.QUEUE_NAME,
+ String queueName = dagAMConf.get(JobContext.QUEUE_NAME,
YarnConfiguration.DEFAULT_QUEUE_NAME);
dagAMConf.setIfUnset(TezConfiguration.TEZ_QUEUE_NAME, queueName);
dagAMConf.setIfUnset(TezConfiguration.TEZ_AM_VIEW_ACLS,
- tezConf.get(MRJobConfig.JOB_ACL_VIEW_JOB, MRJobConfig.DEFAULT_JOB_ACL_VIEW_JOB));
+ dagAMConf.get(MRJobConfig.JOB_ACL_VIEW_JOB, MRJobConfig.DEFAULT_JOB_ACL_VIEW_JOB));
dagAMConf.setIfUnset(TezConfiguration.TEZ_AM_MODIFY_ACLS,
- tezConf.get(MRJobConfig.JOB_ACL_MODIFY_JOB, MRJobConfig.DEFAULT_JOB_ACL_MODIFY_JOB));
+ dagAMConf.get(MRJobConfig.JOB_ACL_MODIFY_JOB, MRJobConfig.DEFAULT_JOB_ACL_MODIFY_JOB));
// Hardcoding at AM level instead of setting per vertex till TEZ-2710 is available
dagAMConf.setIfUnset(TezConfiguration.TEZ_TASK_SCALE_MEMORY_RESERVE_FRACTION, "0.5");
removeUnwantedSettings(dagAMConf, true);
- return dagAMConf;
}
/**
@@ -263,6 +253,14 @@ public class MRToTezHelper {
JobControlCompiler.configureCompression(tezConf);
convertMRToTezConf(tezConf, mrConf, DeprecatedKeys.getMRToTezRuntimeParamMap());
removeUnwantedSettings(tezConf, false);
+
+ // ShuffleVertexManager Plugin settings
+ // DeprecatedKeys.getMRToTezRuntimeParamMap() only translates min and not max
+ String slowStartFraction = mrConf.get(MRJobConfig.COMPLETED_MAPS_FOR_REDUCE_SLOWSTART);
+ if (slowStartFraction != null) {
+ tezConf.setIfUnset(ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_MIN_SRC_FRACTION, slowStartFraction);
+ tezConf.setIfUnset(ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_MAX_SRC_FRACTION, slowStartFraction);
+ }
}
/**
Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/util/TezCompilerUtil.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/util/TezCompilerUtil.java?rev=1783988&r1=1783987&r2=1783988&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/util/TezCompilerUtil.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/tez/util/TezCompilerUtil.java Wed Feb 22 09:43:41 2017
@@ -36,13 +36,14 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezEdgeDescriptor;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezOperPlan;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezOperator;
+import org.apache.pig.backend.hadoop.executionengine.tez.plan.operator.POFRJoinTez;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.operator.POLocalRearrangeTez;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.operator.POStoreTez;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.operator.POValueOutputTez;
import org.apache.pig.backend.hadoop.executionengine.tez.plan.udf.ReadScalarsTez;
+import org.apache.pig.backend.hadoop.executionengine.tez.runtime.HashValuePartitioner;
import org.apache.pig.backend.hadoop.executionengine.tez.runtime.TezInput;
import org.apache.pig.backend.hadoop.executionengine.tez.runtime.TezOutput;
-import org.apache.pig.builtin.RoundRobinPartitioner;
import org.apache.pig.builtin.TOBAG;
import org.apache.pig.data.DataType;
import org.apache.pig.data.TupleFactory;
@@ -198,8 +199,8 @@ public class TezCompilerUtil {
public static boolean isNonPackageInput(String inputKey, TezOperator tezOp) throws PlanException {
try {
- List<TezInput> inputs = PlanHelper.getPhysicalOperators(tezOp.plan, TezInput.class);
- for (TezInput input : inputs) {
+ List<POFRJoinTez> inputs = PlanHelper.getPhysicalOperators(tezOp.plan, POFRJoinTez.class);
+ for (POFRJoinTez input : inputs) {
if (ArrayUtils.contains(input.getTezInputs(), inputKey)) {
return true;
}
@@ -269,7 +270,7 @@ public class TezCompilerUtil {
} else if (dataMovementType == DataMovementType.SCATTER_GATHER) {
edge.outputClassName = UnorderedPartitionedKVOutput.class.getName();
edge.inputClassName = UnorderedKVInput.class.getName();
- edge.partitionerClass = RoundRobinPartitioner.class;
+ edge.partitionerClass = HashValuePartitioner.class;
}
edge.setIntermediateOutputKeyClass(POValueOutputTez.EmptyWritable.class.getName());
edge.setIntermediateOutputValueClass(TUPLE_CLASS);