You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by an...@apache.org on 2013/10/23 00:44:42 UTC
svn commit: r1534836 - in /pig/branches/branch-0.12: CHANGES.txt
src/org/apache/pig/data/DefaultAbstractBag.java
Author: aniket486
Date: Tue Oct 22 22:44:41 2013
New Revision: 1534836
URL: http://svn.apache.org/r1534836
Log:
PIG-3325: Adding a tuple to a bag is slow (dvryaboy via aniket486)
Modified:
pig/branches/branch-0.12/CHANGES.txt
pig/branches/branch-0.12/src/org/apache/pig/data/DefaultAbstractBag.java
Modified: pig/branches/branch-0.12/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/CHANGES.txt?rev=1534836&r1=1534835&r2=1534836&view=diff
==============================================================================
--- pig/branches/branch-0.12/CHANGES.txt (original)
+++ pig/branches/branch-0.12/CHANGES.txt Tue Oct 22 22:44:41 2013
@@ -22,6 +22,8 @@ Release 0.12.1 (unreleased changes)
BUG FIXES
+PIG-3325: Adding a tuple to a bag is slow (dvryaboy via aniket486)
+
PIG-3512: Reducer estimater is broken by PIG-3497 (daijy)
PIG-3510: New filter extractor fails with more than one filter statement (aniket486 via cheolsoo)
Modified: pig/branches/branch-0.12/src/org/apache/pig/data/DefaultAbstractBag.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/src/org/apache/pig/data/DefaultAbstractBag.java?rev=1534836&r1=1534835&r2=1534836&view=diff
==============================================================================
--- pig/branches/branch-0.12/src/org/apache/pig/data/DefaultAbstractBag.java (original)
+++ pig/branches/branch-0.12/src/org/apache/pig/data/DefaultAbstractBag.java Tue Oct 22 22:44:41 2013
@@ -65,9 +65,13 @@ public abstract class DefaultAbstractBag
// to run through the disk when people ask.
protected long mSize = 0;
- protected int mLastContentsSize = -1;
+ // Number of tuples to sample per bag, to get an estimate of tuple size
+ private static final int SPILL_SAMPLE_SIZE = 100;
+ private static final int SPILL_SAMPLE_FREQUENCY = 10;
- protected long avgTupleSize = 0;
+ long aggSampleTupleSize = 0;
+
+ int sampled = 0;
private boolean spillableRegistered = false;
@@ -79,6 +83,28 @@ public abstract class DefaultAbstractBag
return mSize;
}
+
+ /**
+ * Sample every SPILL_SAMPLE_FREQUENCYth tuple
+ * until we reach a max of SPILL_SAMPLE_SIZE
+ * to get an estimate of the tuple sizes.
+ */
+ protected void sampleContents() {
+ synchronized (mContents) {
+ Iterator<Tuple> iter = mContents.iterator();
+ for (int i = 0; i < sampled * SPILL_SAMPLE_FREQUENCY && iter.hasNext(); i++) {
+ iter.next();
+ }
+ for (int i = sampled; iter.hasNext() && sampled < SPILL_SAMPLE_SIZE; i++) {
+ Tuple t = iter.next();
+ if (i % SPILL_SAMPLE_FREQUENCY == 0) {
+ aggSampleTupleSize += t.getMemorySize();
+ sampled += 1;
+ }
+ }
+ }
+ }
+
/**
* Add a tuple to the bag.
* @param t tuple to add.
@@ -97,9 +123,12 @@ public abstract class DefaultAbstractBag
* should call this method after every time they add an element.
*/
protected void markSpillableIfNecessary() {
- if (!spillableRegistered && getMemorySize() >= SPILL_REGISTER_THRESHOLD) {
- SpillableMemoryManager.getInstance().registerSpillable(this);
- spillableRegistered = true;
+ if (!spillableRegistered) {
+ long estimate = getMemorySize();
+ if ( estimate >= SPILL_REGISTER_THRESHOLD) {
+ SpillableMemoryManager.getInstance().registerSpillable(this);
+ spillableRegistered = true;
+ }
}
}
@@ -130,40 +159,30 @@ public abstract class DefaultAbstractBag
*/
@Override
public long getMemorySize() {
- int j = 0;
int numInMem = 0;
synchronized (mContents) {
numInMem = mContents.size();
-
// If we've already gotten the estimate
- // and the number of tuples hasn't changed, or was above 100 and
- // is still above 100, we can
+ // and the number of tuples hasn't changed, or was above
+ // the sample size and is still above the sample size, we can
// produce a new estimate without sampling the tuples again.
- if (avgTupleSize != 0 && (mLastContentsSize == numInMem ||
- mLastContentsSize > 100 && numInMem > 100))
- return totalSizeFromAvgTupleSize(avgTupleSize, numInMem);
-
- // Measure only what's in memory, not what's on disk.
- // I can't afford to walk through all the tuples every time the
- // memory manager wants to know if it's time to dump. Just sample
- // the first 100 and see what we get. This may not be 100%
- // accurate, but it's just an estimate anyway.
- Iterator<Tuple> i = mContents.iterator();
- for (j = 0; i.hasNext() && j < 100; j++) {
- avgTupleSize += i.next().getMemorySize();
+ if (sampled != 0 && (sampled == numInMem ||
+ sampled > SPILL_SAMPLE_SIZE && numInMem > SPILL_SAMPLE_SIZE)) {
+ return totalSizeFromAvgTupleSize(aggSampleTupleSize/sampled, numInMem);
+ }
+ sampleContents();
+ int avgTupleSize;
+ if (sampled != 0) {
+ avgTupleSize = (int) (aggSampleTupleSize / sampled);
+ } else {
+ avgTupleSize = 0;
}
- }
- mLastContentsSize = numInMem;
- if (j != 0) {
- avgTupleSize /= j;
- } else {
- avgTupleSize = 0;
- }
- return totalSizeFromAvgTupleSize(avgTupleSize, numInMem);
+ return totalSizeFromAvgTupleSize(avgTupleSize, numInMem);
}
+ }
private long totalSizeFromAvgTupleSize(long avgTupleSize, int numInMem) {
long used = avgTupleSize * numInMem;
@@ -227,6 +246,9 @@ public abstract class DefaultAbstractBag
mSpillFiles.clear();
}
mSize = 0;
+ aggSampleTupleSize = 0;
+ sampled = 0;
+ // not changing spillableRegistered -- clear doesn't change that.
}
}