You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ga...@apache.org on 2010/12/06 22:44:49 UTC
svn commit: r1042804 - in /pig/trunk: ./ src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/
src/org/apache/pig/newplan/logical/expre...
Author: gates
Date: Mon Dec 6 21:44:49 2010
New Revision: 1042804
URL: http://svn.apache.org/viewvc?rev=1042804&view=rev
Log:
PIG-1752: Enable UDFs to indicate files to load into the Distributed Cache.
Modified:
pig/trunk/CHANGES.txt
pig/trunk/src/org/apache/pig/EvalFunc.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java
pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1042804&r1=1042803&r2=1042804&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Mon Dec 6 21:44:49 2010
@@ -24,6 +24,9 @@ INCOMPATIBLE CHANGES
IMPROVEMENTS
+PIG-1752: Enable UDFs to indicate files to load into the Distributed Cache
+ (gates)
+
PIG-1745: Disable converting bytes loading from BinStorage (daijy)
PIG-1747: pattern match classes for matching patterns in physical plan (thejas)
Modified: pig/trunk/src/org/apache/pig/EvalFunc.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/EvalFunc.java?rev=1042804&r1=1042803&r2=1042804&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/EvalFunc.java (original)
+++ pig/trunk/src/org/apache/pig/EvalFunc.java Mon Dec 6 21:44:49 2010
@@ -266,6 +266,16 @@ public abstract class EvalFunc<T> {
public List<FuncSpec> getArgToFuncMapping() throws FrontendException{
return null;
}
+
+ /**
+ * Allow a UDF to specify a list of files it would like placed in the distributed
+ * cache. These files will be put in the cache for every job the UDF is used in.
+ * The default implementation returns null.
+ * @return A list of files
+ */
+ public List<String> getCacheFiles() {
+ return null;
+ }
public PigLogger getPigLogger() {
return pigLogger;
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java?rev=1042804&r1=1042803&r2=1042804&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java Mon Dec 6 21:44:49 2010
@@ -59,6 +59,7 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFRJoin;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeCogroup;
@@ -503,6 +504,10 @@ public class JobControlCompiler{
// serialized
setupDistributedCacheForJoin(mro, pigContext, conf);
+ // Search to see if we have any UDFs that need to pack things into the
+ // distrubted cache.
+ setupDistributedCacheForUdfs(mro, pigContext, conf);
+
POPackage pack = null;
if(mro.reducePlan.isEmpty()){
//MapOnly Job
@@ -1053,6 +1058,13 @@ public class JobControlCompiler{
.visit();
}
+ private void setupDistributedCacheForUdfs(MapReduceOper mro,
+ PigContext pigContext,
+ Configuration conf) throws IOException {
+ new UdfDistributedCacheVisitor(mro.mapPlan, pigContext, conf).visit();
+ new UdfDistributedCacheVisitor(mro.reducePlan, pigContext, conf).visit();
+ }
+
private static void setupDistributedCache(PigContext pigContext,
Configuration conf,
Properties properties, String key,
@@ -1253,5 +1265,39 @@ public class JobControlCompiler{
}
}
}
-
+
+ private static class UdfDistributedCacheVisitor extends PhyPlanVisitor {
+
+ private PigContext pigContext = null;
+ private Configuration conf = null;
+
+ public UdfDistributedCacheVisitor(PhysicalPlan plan,
+ PigContext pigContext,
+ Configuration conf) {
+ super(plan, new DepthFirstWalker<PhysicalOperator, PhysicalPlan>(
+ plan));
+ this.pigContext = pigContext;
+ this.conf = conf;
+ }
+
+ @Override
+ public void visitUserFunc(POUserFunc func) throws VisitorException {
+
+ // XXX Hadoop currently doesn't support distributed cache in local mode.
+ // This line will be removed after the support is added
+ if (pigContext.getExecType() == ExecType.LOCAL) return;
+
+ // set up distributed cache for files indicated by the UDF
+ String[] files = func.getCacheFiles();
+ if (files == null) return;
+
+ try {
+ setupDistributedCache(pigContext, conf, files, false);
+ } catch (IOException e) {
+ String msg = "Internal error. Distributed cache could not " +
+ "be set up for the requested files";
+ throw new VisitorException(msg, e);
+ }
+ }
+ }
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java?rev=1042804&r1=1042803&r2=1042804&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java Mon Dec 6 21:44:49 2010
@@ -57,6 +57,7 @@ public class POUserFunc extends Expressi
*/
private static final long serialVersionUID = 1L;
transient EvalFunc func;
+ transient private String[] cacheFiles = null;
transient private final Log log = LogFactory.getLog(getClass());
FuncSpec funcSpec;
@@ -432,6 +433,14 @@ public class POUserFunc extends Expressi
return funcSpec;
}
+ public String[] getCacheFiles() {
+ return cacheFiles;
+ }
+
+ public void setCacheFiles(String[] cf) {
+ cacheFiles = cf;
+ }
+
public boolean combinable() {
return (func instanceof Algebraic);
}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java?rev=1042804&r1=1042803&r2=1042804&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java Mon Dec 6 21:44:49 2010
@@ -490,6 +490,10 @@ public class ExpToPhyTranslationVisitor
p = new POUserFunc(new OperatorKey(DEFAULT_SCOPE, nodeGen
.getNextNodeId(DEFAULT_SCOPE)), -1,
null, op.getFuncSpec(), (EvalFunc) f);
+ List<String> cacheFiles = ((EvalFunc)f).getCacheFiles();
+ if (cacheFiles != null) {
+ ((POUserFunc)p).setCacheFiles(cacheFiles.toArray(new String[cacheFiles.size()]));
+ }
} else {
p = new POUserComparisonFunc(new OperatorKey(DEFAULT_SCOPE, nodeGen
.getNextNodeId(DEFAULT_SCOPE)), -1,