svn commit: r1518788 - in /pig/trunk: CHANGES.txt src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java src/org/apache/pig/tools/pigstats/ScriptState.java

[da...@apache.org]

Author: daijy
Date: Thu Aug 29 20:02:01 2013
New Revision: 1518788

URL: http://svn.apache.org/r1518788
Log:
PIG-3048: Add mapreduce workflow information to job configuration

Modified:
    pig/trunk/CHANGES.txt
    pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
    pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java

Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1518788&r1=1518787&r2=1518788&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Thu Aug 29 20:02:01 2013
@@ -28,6 +28,8 @@ PIG-3174: Remove rpm and deb artifacts f
 
 IMPROVEMENTS
 
+PIG-3048: Add mapreduce workflow information to job configuration (billie.rinaldi via daijy)
+
 PIG-3436: Make pigmix run with Hadoop2 (rohini)
 
 PIG-3424: Package import list should consider class name as is first even if -Dudf.import.list is passed (rohini)

Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java?rev=1518788&r1=1518787&r2=1518788&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java Thu Aug 29 20:02:01 2013
@@ -154,6 +154,8 @@ public class MapReduceLauncher extends L
         JobClient jobClient = new JobClient(exe.getJobConf());
         
         JobControlCompiler jcc = new JobControlCompiler(pc, conf);
+
+        ScriptState.get().addWorkflowAdjacenciesToConf(mrp, conf);
         
         // start collecting statistics
         PigStatsUtil.startCollection(pc, jobClient, jcc, mrp); 

Modified: pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java?rev=1518788&r1=1518787&r2=1518788&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java (original)
+++ pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java Thu Aug 29 20:02:01 2013
@@ -333,6 +333,23 @@ public class ScriptState {
         setPigFeature(mro, conf);
 
         setJobParents(mro, conf);
+
+        conf.set("mapreduce.workflow.id", "pig_" + id);
+        conf.set("mapreduce.workflow.name", (getFileName() != null)?getFileName():"default");
+        conf.set("mapreduce.workflow.node.name", mro.getOperatorKey().toString());
+    }
+
+    public void addWorkflowAdjacenciesToConf(MROperPlan mrop, Configuration conf) {
+        for (MapReduceOper source : mrop) {
+            List<String> targets = new ArrayList<String>();
+            if (mrop.getSuccessors(source) != null) {
+                for (MapReduceOper target : mrop.getSuccessors(source)) {
+                    targets.add(target.getOperatorKey().toString());
+                }
+            }
+            String[] s = new String[targets.size()];
+            conf.setStrings("mapreduce.workflow.adjacency." + source.getOperatorKey().toString(), targets.toArray(s));
+        }
     }
 
     public void setScript(File file) {