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/29 01:14:30 UTC
svn commit: r1536564 - in /pig/branches/branch-0.12: ./ conf/
src/docs/src/documentation/content/xdocs/ src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/hadoop/executionengine/util/ src/or...
Author: aniket486
Date: Tue Oct 29 00:14:29 2013
New Revision: 1536564
URL: http://svn.apache.org/r1536564
Log:
PIG-3047: Check the size of a relation before adding it to distributed cache in Replicated join (aniket486)
Modified:
pig/branches/branch-0.12/CHANGES.txt
pig/branches/branch-0.12/conf/pig.properties
pig/branches/branch-0.12/src/docs/src/documentation/content/xdocs/perf.xml
pig/branches/branch-0.12/src/org/apache/pig/PigConfiguration.java
pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java
pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java
pig/branches/branch-0.12/src/org/apache/pig/impl/util/Utils.java
pig/branches/branch-0.12/test/org/apache/pig/test/PigStorageWithStatistics.java
pig/branches/branch-0.12/test/org/apache/pig/test/TestFRJoin2.java
Modified: pig/branches/branch-0.12/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/CHANGES.txt?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/CHANGES.txt (original)
+++ pig/branches/branch-0.12/CHANGES.txt Tue Oct 29 00:14:29 2013
@@ -22,6 +22,8 @@ Release 0.12.1 (unreleased changes)
IMPROVEMENTS
+PIG-3047: Check the size of a relation before adding it to distributed cache in Replicated join (aniket486)
+
PIG-3480: TFile-based tmpfile compression crashes in some cases (dvryaboy via aniket486)
BUG FIXES
Modified: pig/branches/branch-0.12/conf/pig.properties
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/conf/pig.properties?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/conf/pig.properties (original)
+++ pig/branches/branch-0.12/conf/pig.properties Tue Oct 29 00:14:29 2013
@@ -84,7 +84,11 @@
#using more counters than hadoop configured limit
#pig.disable.counter=true
-# Use this option to turn on UDF timers. This will cause two
+# By default, pig will allow 1GB of data to be replicated using
+# the distributed cache when doing fragment-replicated join.
+# pig.join.replicated.max.bytes=1000000000
+
+# Use this option to turn on UDF timers. This will cause two
# counters to be tracked for every UDF and LoadFunc in your script:
# approx_microsecs measures approximate time spent inside a UDF
# approx_invocations reports the approximate number of times the UDF was invoked
Modified: pig/branches/branch-0.12/src/docs/src/documentation/content/xdocs/perf.xml
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/src/docs/src/documentation/content/xdocs/perf.xml?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/src/docs/src/documentation/content/xdocs/perf.xml (original)
+++ pig/branches/branch-0.12/src/docs/src/documentation/content/xdocs/perf.xml Tue Oct 29 00:14:29 2013
@@ -1084,6 +1084,7 @@ C = JOIN big BY b1, tiny BY t1, mini BY
<p>Fragment replicate joins are experimental; we don't have a strong sense of how small the small relation must be to fit
into memory. In our tests with a simple query that involves just a JOIN, a relation of up to 100 M can be used if the process overall
gets 1 GB of memory. Please share your observations and experience with us.</p>
+<p>In order to avoid replicated joins on large relations, we fail if size of relation(s) to be replicated (in bytes) is greater than pig.join.replicated.max.bytes (default = 1GB).</p>
</section>
</section>
<!-- END FRAGMENT REPLICATE JOINS-->
Modified: pig/branches/branch-0.12/src/org/apache/pig/PigConfiguration.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/src/org/apache/pig/PigConfiguration.java?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/src/org/apache/pig/PigConfiguration.java (original)
+++ pig/branches/branch-0.12/src/org/apache/pig/PigConfiguration.java Tue Oct 29 00:14:29 2013
@@ -117,4 +117,10 @@ public class PigConfiguration {
* TFileStorage only support gzip and lzo.
*/
public static final String PIG_TEMP_FILE_COMPRESSION_CODEC = "pig.tmpfilecompression.codec";
+
+ /**
+ * This key used to control the maximum size loaded into
+ * the distributed cache when doing fragment-replicated join
+ */
+ public static final String PIG_JOIN_REPLICATED_MAX_BYTES = "pig.join.replicated.max.bytes";
}
Modified: pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java (original)
+++ pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java Tue Oct 29 00:14:29 2013
@@ -17,6 +17,9 @@
*/
package org.apache.pig.backend.hadoop.executionengine.mapReduceLayer;
+import java.io.IOException;
+import java.util.List;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -29,11 +32,8 @@ import org.apache.pig.LoadMetadata;
import org.apache.pig.ResourceStatistics;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
+import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil;
import org.apache.pig.impl.util.UriUtil;
-import org.apache.pig.impl.util.Utils;
-
-import java.io.IOException;
-import java.util.List;
/**
* Class that estimates the number of reducers based on input size.
@@ -116,7 +116,7 @@ public class InputSizeReducerEstimator i
FileStatus[] status = fs.globStatus(path);
if (status != null) {
for (FileStatus s : status) {
- totalInputFileSize += Utils.getPathLength(fs, s);
+ totalInputFileSize += MapRedUtil.getPathLength(fs, s);
foundSize = true;
}
}
Modified: pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java (original)
+++ pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java Tue Oct 29 00:14:29 2013
@@ -55,6 +55,7 @@ import org.apache.hadoop.mapred.jobcontr
import org.apache.pig.ComparisonFunc;
import org.apache.pig.ExecType;
import org.apache.pig.LoadFunc;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.PigException;
import org.apache.pig.StoreFuncInterface;
import org.apache.pig.backend.executionengine.ExecException;
@@ -76,6 +77,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
import org.apache.pig.backend.hadoop.executionengine.shims.HadoopShims;
+import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataType;
import org.apache.pig.data.SchemaTupleFrontend;
@@ -1551,24 +1553,39 @@ public class JobControlCompiler{
FileSpec[] newReplFiles = new FileSpec[replFiles.length];
// the first input is not replicated
- for (int i = 0; i < replFiles.length; i++) {
- // ignore fragmented file
- String symlink = "";
- if (i != join.getFragment()) {
- symlink = "pigrepl_" + join.getOperatorKey().toString() + "_"
- + Integer.toString(System.identityHashCode(replFiles[i].getFileName()))
- + "_" + Long.toString(System.currentTimeMillis())
- + "_" + i;
- replicatedPath.add(replFiles[i].getFileName() + "#"
- + symlink);
+ long sizeOfReplicatedInputs = 0;
+ try {
+ for (int i = 0; i < replFiles.length; i++) {
+ // ignore fragmented file
+ String symlink = "";
+ if (i != join.getFragment()) {
+ symlink = "pigrepl_" + join.getOperatorKey().toString() + "_"
+ + Integer.toString(System.identityHashCode(
+ replFiles[i].getFileName()))
+ + "_" + Long.toString(System.currentTimeMillis())
+ + "_" + i;
+ replicatedPath.add(replFiles[i].getFileName() + "#"
+ + symlink);
+
+ Path path = new Path(replFiles[i].getFileName());
+ FileSystem fs = path.getFileSystem(conf);
+ sizeOfReplicatedInputs +=
+ MapRedUtil.getPathLength(fs, fs.getFileStatus(path));
+ }
+ newReplFiles[i] = new FileSpec(symlink,
+ (replFiles[i] == null ? null : replFiles[i].getFuncSpec()));
}
- newReplFiles[i] = new FileSpec(symlink,
- (replFiles[i] == null ? null : replFiles[i].getFuncSpec()));
- }
- join.setReplFiles(newReplFiles);
+ join.setReplFiles(newReplFiles);
+
+ String maxSize = pigContext.getProperties().getProperty(
+ PigConfiguration.PIG_JOIN_REPLICATED_MAX_BYTES, "1000000000");
+ if (sizeOfReplicatedInputs > Long.parseLong(maxSize)){
+ throw new VisitorException("Replicated input files size: "
+ + sizeOfReplicatedInputs + " exceeds " +
+ PigConfiguration.PIG_JOIN_REPLICATED_MAX_BYTES + ": " + maxSize);
+ }
- try {
setupDistributedCache(pigContext, conf, replicatedPath
.toArray(new String[0]), false);
} catch (IOException e) {
Modified: pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java (original)
+++ pig/branches/branch-0.12/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java Tue Oct 29 00:14:29 2013
@@ -238,17 +238,36 @@ public class MapRedUtil {
result.add(stat);
}
}
- }
+ }
private static final PathFilter hiddenFileFilter = new PathFilter(){
public boolean accept(Path p){
- String name = p.getName();
- return !name.startsWith("_") && !name.startsWith(".");
+ String name = p.getName();
+ return !name.startsWith("_") && !name.startsWith(".");
}
- };
+ };
+
+ /**
+ * Returns the total number of bytes for this file,
+ * or if a directory all files in the directory.
+ */
+ public static long getPathLength(FileSystem fs, FileStatus status)
+ throws IOException {
+ if (!status.isDir()) {
+ return status.getLen();
+ } else {
+ FileStatus[] children = fs.listStatus(
+ status.getPath(), hiddenFileFilter);
+ long size = 0;
+ for (FileStatus child : children) {
+ size += getPathLength(fs, child);
+ }
+ return size;
+ }
+ }
/* The following codes are for split combination: see PIG-1518
- *
+ *
*/
private static Comparator<Node> nodeComparator = new Comparator<Node>() {
@Override
Modified: pig/branches/branch-0.12/src/org/apache/pig/impl/util/Utils.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/src/org/apache/pig/impl/util/Utils.java?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/src/org/apache/pig/impl/util/Utils.java (original)
+++ pig/branches/branch-0.12/src/org/apache/pig/impl/util/Utils.java Tue Oct 29 00:14:29 2013
@@ -34,7 +34,6 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.GzipCodec;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
@@ -462,21 +461,4 @@ public class Utils {
return in;
}
}
-
- /**
- * Returns the total number of bytes for this file, or if a file all files in the directory.
- */
- public static long getPathLength(FileSystem fs, FileStatus status) throws IOException {
- if (!status.isDir()) {
- return status.getLen();
- } else {
- FileStatus[] children = fs.listStatus(status.getPath());
- long size = 0;
- for (FileStatus child : children) {
- size += getPathLength(fs, child);
- }
- return size;
- }
- }
-
}
Modified: pig/branches/branch-0.12/test/org/apache/pig/test/PigStorageWithStatistics.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/test/org/apache/pig/test/PigStorageWithStatistics.java?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/test/org/apache/pig/test/PigStorageWithStatistics.java (original)
+++ pig/branches/branch-0.12/test/org/apache/pig/test/PigStorageWithStatistics.java Tue Oct 29 00:14:29 2013
@@ -26,8 +26,8 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.pig.ResourceStatistics;
+import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil;
import org.apache.pig.builtin.PigStorage;
-import org.apache.pig.impl.util.Utils;
public class PigStorageWithStatistics extends PigStorage {
private String loc = null;
@@ -58,7 +58,7 @@ public class PigStorageWithStatistics ex
FileStatus[] status = fs.globStatus(path);
if (status != null) {
for (FileStatus s : status) {
- inputBytes += Utils.getPathLength(fs, s);
+ inputBytes += MapRedUtil.getPathLength(fs, s);
}
}
}
Modified: pig/branches/branch-0.12/test/org/apache/pig/test/TestFRJoin2.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.12/test/org/apache/pig/test/TestFRJoin2.java?rev=1536564&r1=1536563&r2=1536564&view=diff
==============================================================================
--- pig/branches/branch-0.12/test/org/apache/pig/test/TestFRJoin2.java (original)
+++ pig/branches/branch-0.12/test/org/apache/pig/test/TestFRJoin2.java Tue Oct 29 00:14:29 2013
@@ -396,7 +396,39 @@ public class TestFRJoin2 {
assertEquals(3, PigStats.get().getJobGraph().size());
}
assertEquals(dbfrj.size(), dbshj.size());
- assertEquals(true, TestHelper.compareBags(dbfrj, dbshj));
+ assertEquals(true, TestHelper.compareBags(dbfrj, dbshj));
+ }
+
+ @Test
+ public void testTooBigReplicatedFile() throws Exception {
+ PigServer pigServer = new PigServer(
+ ExecType.MAPREDUCE, cluster.getProperties());
+
+ pigServer.registerQuery("A = LOAD '" + INPUT_DIR + "' as (x:int,y:int);");
+ pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (x:int,y:int);");
+ pigServer.registerQuery("C = group B all parallel 5;");
+ pigServer.registerQuery("C = foreach C generate MAX(B.x) as x;");
+ pigServer.registerQuery("D = join A by x, B by x, C by x using 'repl';");
+ {
+ // When the replicated input sizes=(12 + 5) is bigger than
+ // pig.join.replicated.max.bytes=16, we throw exception
+ try {
+ pigServer.getPigContext().getProperties().setProperty(
+ PigConfiguration.PIG_JOIN_REPLICATED_MAX_BYTES,
+ String.valueOf(16));
+ pigServer.openIterator("D");
+ Assert.fail();
+ } catch (FrontendException e) {
+ assertEquals("Internal error. Distributed cache could" +
+ " not be set up for the replicated files",
+ e.getCause().getCause().getCause().getMessage());
+ }
+
+ // If we increase the size to 17, it should work
+ pigServer.getPigContext().getProperties().setProperty(
+ PigConfiguration.PIG_JOIN_REPLICATED_MAX_BYTES,
+ String.valueOf(17));
+ pigServer.openIterator("D");
+ }
}
-
}