You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2009/07/13 20:58:31 UTC
svn commit: r793660 - in /hadoop/pig/trunk: ./
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/impl/ test/org/apache/pig/test/
Author: daijy
Date: Mon Jul 13 18:58:31 2009
New Revision: 793660
URL: http://svn.apache.org/viewvc?rev=793660&view=rev
Log:
PIG-883: udf import list does not send to the backend
Modified:
hadoop/pig/trunk/CHANGES.txt
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java
hadoop/pig/trunk/src/org/apache/pig/impl/PigContext.java
hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java
Modified: hadoop/pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/CHANGES.txt (original)
+++ hadoop/pig/trunk/CHANGES.txt Mon Jul 13 18:58:31 2009
@@ -40,6 +40,8 @@
BUG FIXES
+ PIG-883: udf import list does not send to the backend (daijy)
+
PIG-881: Pig should ship load udfs to the backend (daijy)
PIG-876: limit changes order of order-by to ascending (daijy)
Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java Mon Jul 13 18:58:31 2009
@@ -321,6 +321,7 @@
jobConf.set("pig.inputs", ObjectSerializer.serialize(inp));
jobConf.set("pig.inpTargets", ObjectSerializer.serialize(inpTargets));
jobConf.set("pig.pigContext", ObjectSerializer.serialize(pigContext));
+ jobConf.set("udf.import.list", ObjectSerializer.serialize(PigContext.getPackageImportList()));
// this is for unit tests since some don't create PigServer
if (pigContext.getProperties().getProperty(PigContext.JOB_NAME) != null)
jobConf.setJobName(pigContext.getProperties().getProperty(PigContext.JOB_NAME));
Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java Mon Jul 13 18:58:31 2009
@@ -19,6 +19,7 @@
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -94,6 +95,7 @@
super.configure(jConf);
sJobConf = jConf;
try {
+ PigContext.setPackageImportList((ArrayList<String>)ObjectSerializer.deserialize(jConf.get("udf.import.list")));
cp = (PhysicalPlan) ObjectSerializer.deserialize(jConf
.get("pig.combinePlan"));
pack = (POPackage)ObjectSerializer.deserialize(jConf.get("pig.combine.package"));
Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java Mon Jul 13 18:58:31 2009
@@ -190,6 +190,7 @@
.deserialize(job.get("pig.inpTargets"));
pigContext = (PigContext) ObjectSerializer.deserialize(job
.get("pig.pigContext"));
+ PigContext.setPackageImportList((ArrayList<String>)ObjectSerializer.deserialize(job.get("udf.import.list")));
} catch (Exception e) {
int errCode = 2094;
String msg = "Unable to deserialize object.";
Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java Mon Jul 13 18:58:31 2009
@@ -146,6 +146,7 @@
SpillableMemoryManager.configure(ConfigurationUtil.toProperties(job));
PigMapReduce.sJobConf = job;
try {
+ PigContext.setPackageImportList((ArrayList<String>)ObjectSerializer.deserialize(job.get("udf.import.list")));
mp = (PhysicalPlan) ObjectSerializer.deserialize(
job.get("pig.mapPlan"));
stores = PlanHelper.getStores(mp);
Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java Mon Jul 13 18:58:31 2009
@@ -19,6 +19,7 @@
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -180,6 +181,7 @@
SpillableMemoryManager.configure(ConfigurationUtil.toProperties(jConf));
sJobConf = jConf;
try {
+ PigContext.setPackageImportList((ArrayList<String>)ObjectSerializer.deserialize(jConf.get("udf.import.list")));
rp = (PhysicalPlan) ObjectSerializer.deserialize(jConf
.get("pig.reducePlan"));
stores = PlanHelper.getStores(rp);
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/PigContext.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/PigContext.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/PigContext.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/PigContext.java Mon Jul 13 18:58:31 2009
@@ -633,4 +633,8 @@
{
return packageImportList;
}
+ public static void setPackageImportList(ArrayList<String> list)
+ {
+ packageImportList = list;
+ }
}
Modified: hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java (original)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java Mon Jul 13 18:58:31 2009
@@ -23,16 +23,22 @@
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
+import java.io.PrintStream;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import java.util.Properties;
+import java.util.Random;
import junit.framework.TestCase;
import org.apache.hadoop.mapred.FileAlreadyExistsException;
+import org.apache.pig.EvalFunc;
import org.apache.pig.ExecType;
import org.apache.pig.PigServer;
+import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.io.FileLocalizer;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -45,6 +51,7 @@
private File input;
private PigContext pigContext;
+ MiniCluster cluster = MiniCluster.buildCluster();
@Before
@Override
@@ -106,8 +113,14 @@
File udf1Dir = new File(tmpDir.getAbsolutePath()+FILE_SEPARATOR+"com"+FILE_SEPARATOR+"xxx"+FILE_SEPARATOR+"udf1");
udf1Dir.mkdirs();
File udf1JavaSrc = new File(udf1Dir.getAbsolutePath()+FILE_SEPARATOR+"TestUDF.java");
- String udf1Src = new String("package com.xxx.udf1;\n" +
- "public class TestUDF {}\n");
+ String udf1Src = new String("package com.xxx.udf1;\n"+
+ "import java.io.IOException;\n"+
+ "import org.apache.pig.EvalFunc;\n"+
+ "import org.apache.pig.data.Tuple;\n"+
+ "public class TestUDF extends EvalFunc<Integer>{\n"+
+ "public Integer exec(Tuple input) throws IOException {\n"+
+ "return 1;}\n"+
+ "}");
// generate java file
FileOutputStream outStream =
@@ -119,7 +132,7 @@
// compile
int status;
- status = Util.executeShellCommand("javac " + udf1JavaSrc);
+ status = Util.executeShellCommand("javac -cp "+System.getProperty("java.class.path") + " " + udf1JavaSrc);
// generate jar file
String jarName = "TestUDFJar1.jar";
@@ -142,6 +155,30 @@
Object udf = PigContext.instantiateFuncFromSpec("TestUDF");
assertTrue(udf.getClass().toString().endsWith("com.xxx.udf1.TestUDF"));
+ int LOOP_COUNT = 40;
+ File tmpFile = File.createTempFile("test", "txt");
+ PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
+ Random r = new Random(1);
+ int rand;
+ for(int i = 0; i < LOOP_COUNT; i++) {
+ rand = r.nextInt(100);
+ ps.println(rand);
+ }
+ ps.close();
+
+ FileLocalizer.deleteTempFiles();
+ PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer.registerQuery("A = LOAD '" + Util.generateURI(tmpFile.toString()) + "' AS (num:chararray);");
+ pigServer.registerQuery("B = foreach A generate TestUDF(num);");
+ Iterator<Tuple> iter = pigServer.openIterator("B");
+ if(!iter.hasNext()) fail("No output found");
+ int numIdentity = 0;
+ while(iter.hasNext()){
+ Tuple t = iter.next();
+ assertTrue(t.get(0) instanceof Integer);
+ assertTrue((Integer)t.get(0) == 1);
+ }
+
Util.deleteDirectory(tempDir);
}