You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ro...@apache.org on 2014/10/23 23:38:37 UTC
svn commit: r1633955 - in /pig/trunk: ./ src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/
src/org/apache/pig/builtin/ src/org/apache/pig/data/
test/org/apache/pig/test/
Author: rohini
Date: Thu Oct 23 21:38:37 2014
New Revision: 1633955
URL: http://svn.apache.org/r1633955
Log:
PIG-3456: Reduce threadlocal conf access in backend for each record (rohini)
Modified:
pig/trunk/CHANGES.txt
pig/trunk/src/org/apache/pig/PigConfiguration.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/CombinerPackager.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCollectedGroup.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODistinct.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPartialAgg.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSort.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/Packager.java
pig/trunk/src/org/apache/pig/builtin/Distinct.java
pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java
pig/trunk/src/org/apache/pig/data/DefaultTuple.java
pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java
pig/trunk/src/org/apache/pig/data/SelfSpillBag.java
pig/trunk/test/org/apache/pig/test/TestPOPartialAgg.java
pig/trunk/test/org/apache/pig/test/TestTuple.java
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Thu Oct 23 21:38:37 2014
@@ -34,6 +34,8 @@ INCOMPATIBLE CHANGES
IMPROVEMENTS
+PIG-3456: Reduce threadlocal conf access in backend for each record (rohini)
+
PIG-3861: duplicate jars get added to distributed cache (chitnis via rohini)
PIG-4039: New interface for resetting static variables for jvm reuse (rohini)
Modified: pig/trunk/src/org/apache/pig/PigConfiguration.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/PigConfiguration.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/PigConfiguration.java (original)
+++ pig/trunk/src/org/apache/pig/PigConfiguration.java Thu Oct 23 21:38:37 2014
@@ -33,7 +33,14 @@ public class PigConfiguration {
* Controls the fraction of total memory that is allowed to be used by
* cached bags. Default is 0.2.
*/
- public static final String PROP_CACHEDBAG_MEMUSAGE = "pig.cachedbag.memusage";
+ public static final String PIG_CACHEDBAG_MEMUSAGE = "pig.cachedbag.memusage";
+
+ /**
+ * Configurations for specifying alternate implementations for cached bags
+ */
+ public static final String PIG_CACHEDBAG_TYPE = "pig.cachedbag.type";
+ public static final String PIG_CACHEDBAG_DISTINCT_TYPE = "pig.cachedbag.distinct.type";
+ public static final String PIG_CACHEDBAG_SORT_TYPE = "pig.cachedbag.sort.type";
public static final String ACCUMULATIVE_BATCHSIZE = "pig.accumulative.batchsize";
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/CombinerPackager.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/CombinerPackager.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/CombinerPackager.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/CombinerPackager.java Thu Oct 23 21:38:37 2014
@@ -20,6 +20,7 @@ package org.apache.pig.backend.hadoop.ex
import java.util.Arrays;
import java.util.Map;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
@@ -48,6 +49,9 @@ public class CombinerPackager extends Pa
private Map<Integer, Integer> keyLookup;
private int numBags;
+
+ private transient boolean initialized;
+ private transient boolean useDefaultBag;
/**
* A new POPostCombinePackage will be constructed as a near clone of the
@@ -91,15 +95,16 @@ public class CombinerPackager extends Pa
}
private DataBag createDataBag(int numBags) {
- String bagType = null;
- if (PigMapReduce.sJobConfInternal.get() != null) {
- bagType = PigMapReduce.sJobConfInternal.get().get("pig.cachedbag.type");
- }
-
- if (bagType != null && bagType.equalsIgnoreCase("default")) {
- return new NonSpillableDataBag();
+ if (!initialized) {
+ initialized = true;
+ if (PigMapReduce.sJobConfInternal.get() != null) {
+ String bagType = PigMapReduce.sJobConfInternal.get().get(PigConfiguration.PIG_CACHEDBAG_TYPE);
+ if (bagType != null && bagType.equalsIgnoreCase("default")) {
+ useDefaultBag = true;
+ }
+ }
}
- return new InternalCachedBag(numBags);
+ return useDefaultBag ? new NonSpillableDataBag() : new InternalCachedBag(numBags);
}
@Override
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCollectedGroup.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCollectedGroup.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCollectedGroup.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCollectedGroup.java Thu Oct 23 21:38:37 2014
@@ -19,8 +19,8 @@ package org.apache.pig.backend.hadoop.ex
import java.util.ArrayList;
import java.util.List;
-import java.util.Iterator;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
@@ -38,9 +38,6 @@ import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.impl.plan.PlanException;
import org.apache.pig.impl.plan.VisitorException;
-import org.apache.pig.pen.util.ExampleTuple;
-import org.apache.pig.pen.util.LineageTracer;
-import org.apache.pig.impl.util.IdentityHashSet;
/**
* The collected group operator is a special operator used when users give
@@ -71,7 +68,7 @@ public class POCollectedGroup extends Ph
private Object prevKey = null;
- private boolean useDefaultBag = false;
+ private transient boolean useDefaultBag;
public POCollectedGroup(OperatorKey k) {
this(k, -1, null);
@@ -172,7 +169,7 @@ public class POCollectedGroup extends Ph
if (prevKey == null && outputBag == null) {
if (PigMapReduce.sJobConfInternal.get() != null) {
- String bagType = PigMapReduce.sJobConfInternal.get().get("pig.cachedbag.type");
+ String bagType = PigMapReduce.sJobConfInternal.get().get(PigConfiguration.PIG_CACHEDBAG_TYPE);
if (bagType != null && bagType.equalsIgnoreCase("default")) {
useDefaultBag = true;
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODistinct.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODistinct.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODistinct.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODistinct.java Thu Oct 23 21:38:37 2014
@@ -23,6 +23,7 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
@@ -42,15 +43,18 @@ import org.apache.pig.impl.plan.VisitorE
* Find the distinct set of tuples in a bag.
* This is a blocking operator. All the input is put in the hashset implemented
* in DistinctDataBag which also provides the other DataBag interfaces.
- *
- *
+ *
+ *
*/
public class PODistinct extends PhysicalOperator implements Cloneable {
private static final Log log = LogFactory.getLog(PODistinct.class);
private static final long serialVersionUID = 1L;
private boolean inputsAccumulated = false;
private DataBag distinctBag = null;
- transient Iterator<Tuple> it;
+
+ private transient boolean initialized;
+ private transient boolean useDefaultBag;
+ private transient Iterator<Tuple> it;
// PIG-3385: Since GlobalRearrange is not used by PODistinct, passing the
// custom partioner through here
@@ -87,17 +91,19 @@ public class PODistinct extends Physical
@Override
public Result getNextTuple() throws ExecException {
if (!inputsAccumulated) {
- // by default, we create InternalSortedBag, unless user configures
+ // by default, we create InternalDistinctBag, unless user configures
// explicitly to use old bag
- String bagType = null;
- if (PigMapReduce.sJobConfInternal.get() != null) {
- bagType = PigMapReduce.sJobConfInternal.get().get("pig.cachedbag.distinct.type");
- }
- if (bagType != null && bagType.equalsIgnoreCase("default")) {
- distinctBag = BagFactory.getInstance().newDistinctBag();
- } else {
- distinctBag = new InternalDistinctBag(3);
- }
+ if (!initialized) {
+ initialized = true;
+ if (PigMapReduce.sJobConfInternal.get() != null) {
+ String bagType = PigMapReduce.sJobConfInternal.get().get(PigConfiguration.PIG_CACHEDBAG_DISTINCT_TYPE);
+ if (bagType != null && bagType.equalsIgnoreCase("default")) {
+ useDefaultBag = true;
+ }
+ }
+ }
+ distinctBag = useDefaultBag ? BagFactory.getInstance().newDistinctBag()
+ : new InternalDistinctBag(3);
Result in = processInput();
while (in.returnStatus != POStatus.STATUS_EOP) {
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPackage.java Thu Oct 23 21:38:37 2014
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
@@ -76,16 +77,14 @@ public class POPackage extends PhysicalO
protected static final BagFactory mBagFactory = BagFactory.getInstance();
protected static final TupleFactory mTupleFactory = TupleFactory.getInstance();
- private boolean firstTime = true;
-
- private boolean useDefaultBag = false;
-
private boolean lastBagReadOnly = true;
protected Packager pkgr;
protected PigNullableWritable keyWritable;
+ private transient boolean initialized;
+ private transient boolean useDefaultBag;
private transient int accumulativeBatchSize;
public POPackage(OperatorKey k) {
@@ -192,11 +191,11 @@ public class POPackage extends PhysicalO
*/
@Override
public Result getNextTuple() throws ExecException {
- if(firstTime){
- firstTime = false;
+ if (!initialized) {
+ initialized = true;
if (PigMapReduce.sJobConfInternal.get() != null) {
String bagType = PigMapReduce.sJobConfInternal.get().get(
- "pig.cachedbag.type");
+ PigConfiguration.PIG_CACHEDBAG_TYPE);
if (bagType != null && bagType.equalsIgnoreCase("default")) {
useDefaultBag = true;
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPartialAgg.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPartialAgg.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPartialAgg.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPartialAgg.java Thu Oct 23 21:38:37 2014
@@ -104,6 +104,8 @@ public class POPartialAgg extends Physic
private transient MemoryLimits memLimits;
private transient boolean initialized = false;
+ private int minOutputReduction;
+ private float percentUsage;
private int firstTierThreshold = FIRST_TIER_THRESHOLD;
private int secondTierThreshold = SECOND_TIER_THRESHOLD;
private int sizeReduction = 1;
@@ -118,8 +120,22 @@ public class POPartialAgg extends Physic
private void init() throws ExecException {
ALL_POPARTS.put(this, null);
- float percent = getPercentUsageFromProp();
- if (percent <= 0) {
+ percentUsage = 0.2F;
+ if (PigMapReduce.sJobConfInternal.get() != null) {
+ String usage = PigMapReduce.sJobConfInternal.get().get(
+ PigConfiguration.PIG_CACHEDBAG_MEMUSAGE);
+ if (usage != null) {
+ percentUsage = Float.parseFloat(usage);
+ }
+ minOutputReduction = PigMapReduce.sJobConfInternal.get().getInt(
+ PigConfiguration.PARTAGG_MINREDUCTION, DEFAULT_MIN_REDUCTION);
+ if (minOutputReduction <= 0) {
+ LOG.info("Specified reduction is < 0 (" + minOutputReduction + "). Using default " +
+ DEFAULT_MIN_REDUCTION);
+ minOutputReduction = DEFAULT_MIN_REDUCTION;
+ }
+ }
+ if (percentUsage <= 0) {
LOG.info("No memory allocated to intermediate memory buffers. Turning off partial aggregation.");
disableMapAgg();
}
@@ -216,8 +232,7 @@ public class POPartialAgg extends Physic
if (!mapAggDisabled()) {
LOG.info("Getting mem limits; considering " + ALL_POPARTS.size() + " POPArtialAgg objects.");
- float percent = getPercentUsageFromProp();
- memLimits = new MemoryLimits(ALL_POPARTS.size(), percent);
+ memLimits = new MemoryLimits(ALL_POPARTS.size(), percentUsage);
int estTotalMem = 0;
int estTuples = 0;
for (Map.Entry<Object, List<Tuple>> entry : rawInputMap.entrySet()) {
@@ -244,12 +259,11 @@ public class POPartialAgg extends Physic
aggregateSecondLevel();
int numAfterReduction = numRecsInProcessedMap + numRecsInRawMap;
LOG.info("After reduction, processed map: " + numRecsInProcessedMap + "; raw map: " + numRecsInRawMap);
- int minReduction = getMinOutputReductionFromProp();
LOG.info("Observed reduction factor: from " + numBeforeReduction +
" to " + numAfterReduction +
" => " + numBeforeReduction / numAfterReduction + ".");
- if ( numBeforeReduction / numAfterReduction < minReduction) {
- LOG.info("Disabling in-memory aggregation, since observed reduction is less than " + minReduction);
+ if ( numBeforeReduction / numAfterReduction < minOutputReduction) {
+ LOG.info("Disabling in-memory aggregation, since observed reduction is less than " + minOutputReduction);
disableMapAgg();
}
sizeReduction = numBeforeReduction / numAfterReduction;
@@ -424,29 +438,6 @@ public class POPartialAgg extends Physic
v.visitPartialAgg(this);
}
- private int getMinOutputReductionFromProp() {
- int minReduction = PigMapReduce.sJobConfInternal.get().getInt(
- PigConfiguration.PARTAGG_MINREDUCTION, DEFAULT_MIN_REDUCTION);
- if (minReduction <= 0) {
- LOG.info("Specified reduction is < 0 (" + minReduction + "). Using default " + DEFAULT_MIN_REDUCTION);
- minReduction = DEFAULT_MIN_REDUCTION;
- }
- return minReduction;
- }
-
- private float getPercentUsageFromProp() {
- float percent = 0.2F;
- if (PigMapReduce.sJobConfInternal.get() != null) {
- String usage = PigMapReduce.sJobConfInternal.get().get(
- PigConfiguration.PROP_CACHEDBAG_MEMUSAGE);
- if (usage != null) {
- percent = Float.parseFloat(usage);
- }
- }
- return percent;
- }
-
-
private Result getResult(ExpressionOperator op) throws ExecException {
Result res;
switch (op.getResultType()) {
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSort.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSort.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSort.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POSort.java Thu Oct 23 21:38:37 2014
@@ -25,6 +25,7 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
@@ -77,7 +78,10 @@ public class POSort extends PhysicalOper
private long limit;
public boolean isUDFComparatorUsed = false;
private DataBag sortedBag;
- transient Iterator<Tuple> it;
+
+ private transient Iterator<Tuple> it;
+ private transient boolean initialized;
+ private transient boolean useDefaultBag;
public POSort(
OperatorKey k,
@@ -256,17 +260,19 @@ public class POSort extends PhysicalOper
if (!inputsAccumulated) {
res = processInput();
+ if (!initialized) {
+ initialized = true;
+ if (PigMapReduce.sJobConfInternal.get() != null) {
+ String bagType = PigMapReduce.sJobConfInternal.get().get(PigConfiguration.PIG_CACHEDBAG_SORT_TYPE);
+ if (bagType != null && bagType.equalsIgnoreCase("default")) {
+ useDefaultBag = true;
+ }
+ }
+ }
// by default, we create InternalSortedBag, unless user configures
- // explicitly to use old bag
- String bagType = null;
- if (PigMapReduce.sJobConfInternal.get() != null) {
- bagType = PigMapReduce.sJobConfInternal.get().get("pig.cachedbag.sort.type");
- }
- if (bagType != null && bagType.equalsIgnoreCase("default")) {
- sortedBag = BagFactory.getInstance().newSortedBag(mComparator);
- } else {
- sortedBag = new InternalSortedBag(3, mComparator);
- }
+ // explicitly to use old bag
+ sortedBag = useDefaultBag ? BagFactory.getInstance().newSortedBag(mComparator)
+ : new InternalSortedBag(3, mComparator);
while (res.returnStatus != POStatus.STATUS_EOP) {
if (res.returnStatus == POStatus.STATUS_ERR) {
@@ -377,6 +383,7 @@ public class POSort extends PhysicalOper
}
+ @Override
public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
if(illustrator != null) {
illustrator.getEquivalenceClasses().get(eqClassIndex).add((Tuple) in);
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/Packager.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/Packager.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/Packager.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/Packager.java Thu Oct 23 21:38:37 2014
@@ -23,6 +23,7 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
@@ -92,8 +93,8 @@ public class Packager implements Illustr
private PackageType pkgType;
- boolean firstTime = true;
- boolean useDefaultBag = false;
+ private transient boolean initialized;
+ private transient boolean useDefaultBag;
protected POPackage parent = null;
@@ -473,10 +474,10 @@ public class Packager implements Illustr
}
public void checkBagType() {
- if(firstTime){
- firstTime = false;
+ if(!initialized){
+ initialized = true;
if (PigMapReduce.sJobConfInternal.get() != null) {
- String bagType = PigMapReduce.sJobConfInternal.get().get("pig.cachedbag.type");
+ String bagType = PigMapReduce.sJobConfInternal.get().get(PigConfiguration.PIG_CACHEDBAG_TYPE);
if (bagType != null && bagType.equalsIgnoreCase("default")) {
useDefaultBag = true;
}
Modified: pig/trunk/src/org/apache/pig/builtin/Distinct.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/builtin/Distinct.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/builtin/Distinct.java (original)
+++ pig/trunk/src/org/apache/pig/builtin/Distinct.java Thu Oct 23 21:38:37 2014
@@ -22,6 +22,9 @@ import java.io.IOException;
import org.apache.pig.Algebraic;
import org.apache.pig.EvalFunc;
+import org.apache.pig.JVMReuseManager;
+import org.apache.pig.PigConfiguration;
+import org.apache.pig.StaticDataCleanup;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.data.BagFactory;
@@ -38,35 +41,35 @@ import org.apache.pig.data.TupleFactory;
*/
public class Distinct extends EvalFunc<DataBag> implements Algebraic {
- private static BagFactory bagFactory = BagFactory.getInstance();
private static TupleFactory tupleFactory = TupleFactory.getInstance();
- /* (non-Javadoc)
- * @see org.apache.pig.EvalFunc#exec(org.apache.pig.data.Tuple)
- */
+ private static boolean initialized = false;
+ private static boolean useDefaultBag = false;
+
+ static {
+ JVMReuseManager.getInstance().registerForStaticDataCleanup(Distinct.class);
+ }
+
+ @StaticDataCleanup
+ public static void staticDataCleanup() {
+ initialized = false;
+ useDefaultBag = false;
+ }
+
@Override
public DataBag exec(Tuple input) throws IOException {
return getDistinct(input);
}
- /* (non-Javadoc)
- * @see org.apache.pig.Algebraic#getFinal()
- */
@Override
public String getFinal() {
return Final.class.getName();
}
- /* (non-Javadoc)
- * @see org.apache.pig.Algebraic#getInitial()
- */
@Override
public String getInitial() {
return Initial.class.getName();
}
- /* (non-Javadoc)
- * @see org.apache.pig.Algebraic#getIntermed()
- */
@Override
public String getIntermed() {
return Intermediate.class.getName();
@@ -74,13 +77,10 @@ public class Distinct extends EvalFunc<
static public class Initial extends EvalFunc<Tuple> {
- /* (non-Javadoc)
- * @see org.apache.pig.EvalFunc#exec(org.apache.pig.data.Tuple)
- */
@Override
public Tuple exec(Tuple input) throws IOException {
// the input has a single field which is a tuple
- // representing the data we want to distinct.
+ // representing the data we want to distinct.
// unwrap, put in a bag and send down
try {
Tuple single = (Tuple)input.get(0);
@@ -94,9 +94,6 @@ public class Distinct extends EvalFunc<
static public class Intermediate extends EvalFunc<Tuple> {
- /* (non-Javadoc)
- * @see org.apache.pig.EvalFunc#exec(org.apache.pig.data.Tuple)
- */
@Override
public Tuple exec(Tuple input) throws IOException {
return tupleFactory.newTuple(getDistinctFromNestedBags(input, this));
@@ -105,30 +102,27 @@ public class Distinct extends EvalFunc<
static public class Final extends EvalFunc<DataBag> {
- /* (non-Javadoc)
- * @see org.apache.pig.EvalFunc#exec(org.apache.pig.data.Tuple)
- */
@Override
public DataBag exec(Tuple input) throws IOException {
return getDistinctFromNestedBags(input, this);
}
}
-
- static private DataBag createDataBag() {
- // by default, we create InternalSortedBag, unless user configures
- // explicitly to use old bag
- String bagType = null;
- if (PigMapReduce.sJobConfInternal.get() != null) {
- bagType = PigMapReduce.sJobConfInternal.get().get("pig.cachedbag.distinct.type");
- }
-
- if (bagType != null && bagType.equalsIgnoreCase("default")) {
- return BagFactory.getInstance().newDistinctBag();
- } else {
- return new InternalDistinctBag(3);
- }
+
+ private static DataBag createDataBag() {
+ if (!initialized) {
+ initialized = true;
+ if (PigMapReduce.sJobConfInternal.get() != null) {
+ String bagType = PigMapReduce.sJobConfInternal.get().get(PigConfiguration.PIG_CACHEDBAG_DISTINCT_TYPE);
+ if (bagType != null && bagType.equalsIgnoreCase("default")) {
+ useDefaultBag = true;
+ }
+ }
+ }
+ // by default, we create InternalDistinctBag, unless user configures
+ // explicitly to use old bag
+ return useDefaultBag ? BagFactory.getInstance().newDistinctBag() : new InternalDistinctBag(3);
}
-
+
static private DataBag getDistinctFromNestedBags(Tuple input, EvalFunc evalFunc) throws IOException {
DataBag result = createDataBag();
long progressCounter = 0;
@@ -144,7 +138,7 @@ public class Distinct extends EvalFunc<
for (Tuple t : (DataBag)tuple.get(0)) {
result.add(t);
++progressCounter;
- if((progressCounter % 1000) == 0){
+ if((progressCounter % 1000) == 0){
evalFunc.progress();
}
}
@@ -154,7 +148,7 @@ public class Distinct extends EvalFunc<
}
return result;
}
-
+
protected DataBag getDistinct(Tuple input) throws IOException {
try {
DataBag inputBg = (DataBag)input.get(0);
Modified: pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java (original)
+++ pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java Thu Oct 23 21:38:37 2014
@@ -187,19 +187,18 @@ public abstract class DefaultAbstractBag
private long totalSizeFromAvgTupleSize(long avgTupleSize, int numInMem) {
long used = avgTupleSize * numInMem;
- // add up the overhead for this object and other object variables
- int bag_fix_size = 8 /* object header */
- + 4 + 8 + 8 /* mLastContentsSize + mMemSize + mSize */
- + 8 + 8 /* mContents ref + mSpillFiles ref*/
- + 4 /* +4 to round it to eight*/
- + 36 /* mContents fixed */
- ;
long mFields_size = roundToEight(4 + numInMem*4); /* mContents fixed + per entry */
// in java hotspot 32bit vm, there seems to be a minimum bag size of 188 bytes
// some of the extra bytes is probably from a minimum size of this array list
mFields_size = Math.max(40, mFields_size);
- used += bag_fix_size + mFields_size;
+ // the fixed overhead for this object and other object variables = 84 bytes
+ // 8 - object header
+ // 4 + 8 + 8 - sampled + aggSampleTupleSize + mSize
+ // 8 + 8 - mContents ref + mSpillFiles ref
+ // 4 - spillableRegistered +4 instead of 1 to round it to eight
+ // 36 - mContents fixed
+ used += 84 + mFields_size;
// add up overhead for mSpillFiles ArrayList, Object[] inside ArrayList,
// object variable inside ArrayList and references to spill files
Modified: pig/trunk/src/org/apache/pig/data/DefaultTuple.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/DefaultTuple.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/DefaultTuple.java (original)
+++ pig/trunk/src/org/apache/pig/data/DefaultTuple.java Thu Oct 23 21:38:37 2014
@@ -45,7 +45,6 @@ import org.apache.pig.impl.util.ObjectSe
*/
public class DefaultTuple extends AbstractTuple {
- protected boolean isNull = false;
private static final long serialVersionUID = 2L;
protected List<Object> mFields;
@@ -165,11 +164,6 @@ public class DefaultTuple extends Abstra
@Override
public long getMemorySize() {
Iterator<Object> i = mFields.iterator();
- // fixed overhead
- long empty_tuple_size = 8 /* tuple object header */
- + 8 /* isNull - but rounded to 8 bytes as total obj size needs to be multiple of 8 */
- + 8 /* mFields reference */
- + 32 /* mFields array list fixed size */;
// rest of the fixed portion of mfields size is accounted within empty_tuple_size
long mfields_var_size = SizeUtil.roundToEight(4 + 4 * mFields.size());
@@ -177,7 +171,11 @@ public class DefaultTuple extends Abstra
// which is probably from the minimum size of this array list
mfields_var_size = Math.max(40, mfields_var_size);
- long sum = empty_tuple_size + mfields_var_size;
+ // fixed overhead = 48 bytes
+ //8 - tuple object header
+ //8 - mFields reference
+ //32 - mFields array list fixed size
+ long sum = 48 + mfields_var_size;
while (i.hasNext()) {
sum += SizeUtil.getPigObjMemSize(i.next());
}
Modified: pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java (original)
+++ pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java Thu Oct 23 21:38:37 2014
@@ -81,7 +81,7 @@ public class InternalDistinctBag extends
if (percent < 0) {
percent = 0.2F;
if (PigMapReduce.sJobConfInternal.get() != null) {
- String usage = PigMapReduce.sJobConfInternal.get().get(PigConfiguration.PROP_CACHEDBAG_MEMUSAGE);
+ String usage = PigMapReduce.sJobConfInternal.get().get(PigConfiguration.PIG_CACHEDBAG_MEMUSAGE);
if (usage != null) {
percent = Float.parseFloat(usage);
}
Modified: pig/trunk/src/org/apache/pig/data/SelfSpillBag.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/SelfSpillBag.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/SelfSpillBag.java (original)
+++ pig/trunk/src/org/apache/pig/data/SelfSpillBag.java Thu Oct 23 21:38:37 2014
@@ -63,7 +63,7 @@ public abstract class SelfSpillBag exten
maxMem = Runtime.getRuntime().maxMemory();
if (PigMapReduce.sJobConfInternal.get() != null) {
String usage = PigMapReduce.sJobConfInternal.get().get(
- PigConfiguration.PROP_CACHEDBAG_MEMUSAGE);
+ PigConfiguration.PIG_CACHEDBAG_MEMUSAGE);
if (usage != null) {
cachedMemUsage = Float.parseFloat(usage);
}
Modified: pig/trunk/test/org/apache/pig/test/TestPOPartialAgg.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestPOPartialAgg.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestPOPartialAgg.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestPOPartialAgg.java Thu Oct 23 21:38:37 2014
@@ -290,7 +290,7 @@ public class TestPOPartialAgg {
PigMapReduce.sJobConfInternal.set(new Configuration());
if (isMapMemEmpty) {
- PigMapReduce.sJobConfInternal.get().set(PigConfiguration.PROP_CACHEDBAG_MEMUSAGE,
+ PigMapReduce.sJobConfInternal.get().set(PigConfiguration.PIG_CACHEDBAG_MEMUSAGE,
"0");
}
Modified: pig/trunk/test/org/apache/pig/test/TestTuple.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestTuple.java?rev=1633955&r1=1633954&r2=1633955&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestTuple.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestTuple.java Thu Oct 23 21:38:37 2014
@@ -81,31 +81,31 @@ public class TestTuple {
public void testEmptyTupleSize() {
Tuple t = mTupleFactory.newTuple();
long size = t.getMemorySize();
- assertEquals("tuple size",size, 96);
+ assertEquals("tuple size", 88, size);
}
-
+
@Test
public void testEmptyBagSize() {
DataBag bag = DefaultBagFactory.getInstance().newDefaultBag();
long size = bag.getMemorySize();
- assertEquals("bag size",size, 124);
+ assertEquals("bag size", 124, size);
}
-
+
@Test
// See PIG-1443
public void testTupleSizeWithString() {
Tuple t = Util.createTuple(new String[] {"1234567", "bar"});
long size = t.getMemorySize();
- assertEquals("tuple size",size, 200);
+ assertEquals("tuple size", 192, size);
}
-
+
@Test
public void testTupleSizeWithByteArrays() {
Tuple t = mTupleFactory.newTuple();
t.append(new DataByteArray("1234567"));
t.append(new DataByteArray("bar"));
long size = t.getMemorySize();
- assertEquals("tuple size",size, 168);
+ assertEquals("tuple size", 160, size);
}
@Test
@@ -114,7 +114,7 @@ public class TestTuple {
t.append(new Double(0.1));
t.append(new Double(2000.10001));
long size = t.getMemorySize();
- assertEquals("tuple size",size, 128);
+ assertEquals("tuple size", 120, size);
}
@Test
@@ -123,27 +123,27 @@ public class TestTuple {
t.append(new Float(0.1F));
t.append(new Float(2000.10001F));
long size = t.getMemorySize();
- assertEquals("tuple size",size, 128);
+ assertEquals("tuple size", 120, size);
}
-
+
@Test
public void testTupleSizeWithLongs() {
Tuple t = mTupleFactory.newTuple();
t.append(new Long(100));
t.append(new Long(2000));
long size = t.getMemorySize();
- assertEquals("tuple size",size, 128);
+ assertEquals("tuple size", 120, size);
}
-
+
@Test
public void testTupleSizeWithBooleans() {
Tuple t = mTupleFactory.newTuple();
t.append(new Boolean(true));
t.append(new Boolean(false));
long size = t.getMemorySize();
- assertEquals("tuple size",size, 128);
- }
-
+ assertEquals("tuple size", 120, size);
+ }
+
@Test
public void testTupleIterator() {
Tuple t = mTupleFactory.newTuple();