MAPREDUCE-5129. Allow tags to JobHistory for deeper analytics. Contributed by Billie Rinaldi.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-1@1465191 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -233,6 +233,9 @@ Release 1.2.0 - unreleased
     HADOOP-9071. Configure ivy log levels for resolve/retrieve.
     (Giridharan Kesavan via suresh)
 
+    MAPREDUCE-5129. Allow tags to JobHistory for deeper analytics. (billie via
+    acmurthy)
+
   BUG FIXES
 
     HADOOP-9451. Fault single-layer config if node group topology is enabled.

src/mapred/org/apache/hadoop/mapred/JobConf.java

@@ -327,6 +327,8 @@ public class JobConf extends Configuration {
   public static final String WORKFLOW_ADJACENCY_PREFIX_PATTERN =
       "^mapreduce\\.workflow\\.adjacency\\..+";
 
+  public static final String WORKFLOW_TAGS = "mapreduce.workflow.tags";
+
   public static final String MAPREDUCE_RECOVER_JOB = 
       "mapreduce.job.restart.recover";
 

src/mapred/org/apache/hadoop/mapred/JobHistory.java

@@ -476,7 +476,8 @@ public class JobHistory {
     SHUFFLE_FINISHED, SORT_FINISHED, COUNTERS, SPLITS, JOB_PRIORITY, HTTP_PORT, 
     TRACKER_NAME, STATE_STRING, VERSION, MAP_COUNTERS, REDUCE_COUNTERS,
     VIEW_JOB, MODIFY_JOB, JOB_QUEUE, FAIL_REASON, LOCALITY, AVATAAR,
-    WORKFLOW_ID, WORKFLOW_NAME, WORKFLOW_NODE_NAME, WORKFLOW_ADJACENCIES
+    WORKFLOW_ID, WORKFLOW_NAME, WORKFLOW_NODE_NAME, WORKFLOW_ADJACENCIES,
+    WORKFLOW_TAGS
   }
 
   /**
@@ -1781,15 +1782,17 @@ public class JobHistory {
                                   Keys.VIEW_JOB, Keys.MODIFY_JOB,
                                   Keys.JOB_QUEUE, Keys.WORKFLOW_ID,
                                   Keys.WORKFLOW_NAME, Keys.WORKFLOW_NODE_NAME,
-                                  Keys.WORKFLOW_ADJACENCIES}, 
+                                  Keys.WORKFLOW_ADJACENCIES,
+                                  Keys.WORKFLOW_TAGS}, 
                        new String[]{jobId.toString(), jobName, user, 
                                     String.valueOf(submitTime) , jobConfPath,
                                     viewJobACL, modifyJobACL,
                                     jobConf.getQueueName(),
-                                    jobConf.get(jobConf.WORKFLOW_ID, ""),
-                                    jobConf.get(jobConf.WORKFLOW_NAME, ""),
-                                    jobConf.get(jobConf.WORKFLOW_NODE_NAME, ""),
-                                    getWorkflowAdjacencies(jobConf)
+                                    jobConf.get(JobConf.WORKFLOW_ID, ""),
+                                    jobConf.get(JobConf.WORKFLOW_NAME, ""),
+                                    jobConf.get(JobConf.WORKFLOW_NODE_NAME, ""),
+                                    getWorkflowAdjacencies(jobConf),
+                                    jobConf.get(JobConf.WORKFLOW_TAGS, ""),
                                     }, 
                                     jobId
                       ); 

src/test/org/apache/hadoop/mapred/TestJobHistory.java

@@ -841,6 +841,18 @@ public class TestJobHistory extends TestCase {
     
     // Validate the job queue name
     assertTrue(jobInfo.getJobQueue().equals(conf.getQueueName()));
+
+    // Validate the workflow properties
+    assertTrue(jobInfo.get(Keys.WORKFLOW_ID).equals(
+        conf.get(JobConf.WORKFLOW_ID, "")));
+    assertTrue(jobInfo.get(Keys.WORKFLOW_NAME).equals(
+        conf.get(JobConf.WORKFLOW_NAME, "")));
+    assertTrue(jobInfo.get(Keys.WORKFLOW_NODE_NAME).equals(
+        conf.get(JobConf.WORKFLOW_NODE_NAME, "")));
+    assertTrue(jobInfo.get(Keys.WORKFLOW_ADJACENCIES).equals(
+        JobHistory.JobInfo.getWorkflowAdjacencies(conf)));
+    assertTrue(jobInfo.get(Keys.WORKFLOW_TAGS).equals(
+        conf.get(JobConf.WORKFLOW_TAGS, "")));
   }
 
   public void testDoneFolderOnHDFS() throws IOException {
@@ -998,7 +1010,21 @@ public class TestJobHistory extends TestCase {
       // no queue admins for default queue
       conf.set(QueueManager.toFullPropertyName(
           "default", QueueACL.ADMINISTER_JOBS.getAclName()), " ");
-      
+
+      // set workflow properties
+      conf.set(JobConf.WORKFLOW_ID, "workflowId1");
+      conf.set(JobConf.WORKFLOW_NAME, "workflowName1");
+      String workflowNodeName = "A";
+      conf.set(JobConf.WORKFLOW_NODE_NAME, workflowNodeName);
+      conf.set(JobConf.WORKFLOW_ADJACENCY_PREFIX_STRING + workflowNodeName,
+          "BC");
+      conf.set(JobConf.WORKFLOW_ADJACENCY_PREFIX_STRING + workflowNodeName,
+          "DEF");
+      conf.set(JobConf.WORKFLOW_ADJACENCY_PREFIX_STRING + "DEF", "G");
+      conf.set(JobConf.WORKFLOW_ADJACENCY_PREFIX_STRING + "Z",
+          workflowNodeName);
+      conf.set(JobConf.WORKFLOW_TAGS, "tag1,tag2");
+
       mr = new MiniMRCluster(2, "file:///", 3, null, null, conf);
 
       // run the TCs

src/tools/org/apache/hadoop/tools/rumen/Job20LineHistoryEventEmitter.java

@@ -82,6 +82,10 @@ public class Job20LineHistoryEventEmitter extends HistoryEventEmitter {
       if (workflowAdjacencies == null) {
         workflowAdjacencies = "";
       }
+      String workflowTags = line.get("WORKFLOW_TAGS");
+      if (workflowTags == null) {
+        workflowTags = "";
+      }
 
       if (submitTime != null) {
         Job20LineHistoryEventEmitter that =
@@ -94,7 +98,7 @@ public class Job20LineHistoryEventEmitter extends HistoryEventEmitter {
         return new JobSubmittedEvent(jobID, jobName,
             user == null ? "nulluser" : user, that.originalSubmitTime,
             jobConf, jobACLs, queueName, workflowId, workflowName,
-            workflowNodeName, workflowAdjacencies);
+            workflowNodeName, workflowAdjacencies, workflowTags);
       }
 
       return null;

src/tools/org/apache/hadoop/tools/rumen/JobSubmittedEvent.java

@@ -42,11 +42,12 @@ public class JobSubmittedEvent implements HistoryEvent {
   private String workflowName;
   private String workflowNodeName;
   private String workflowAdjacencies;
+  private String workflowTags;
 
   /**
    * @deprecated Use
    *             {@link #JobSubmittedEvent(JobID, String, String, long, String,
-   *             Map, String)}
+   *             Map, String, String, String, String, String, String)}
    *             instead.
    */
   @Deprecated
@@ -59,7 +60,7 @@ public class JobSubmittedEvent implements HistoryEvent {
   /**
    * @deprecated Use
    *             {@link #JobSubmittedEvent(JobID, String, String, long, String,
-   *             Map, String)}
+   *             Map, String, String, String, String, String, String)}
    *             instead.
    */
   @Deprecated
@@ -70,6 +71,22 @@ public class JobSubmittedEvent implements HistoryEvent {
         "", "", "", "");
   }
 
+  /**
+   * @deprecated Use
+   *             {@link #JobSubmittedEvent(JobID, String, String, long, String,
+   *             Map, String, String, String, String, String, String)}
+   *             instead.
+   */
+  @Deprecated
+  public JobSubmittedEvent(JobID id, String jobName, String userName,
+      long submitTime, String jobConfPath,
+      Map<JobACL, AccessControlList> jobACLs, String queue,
+      String workflowId, String workflowName, String workflowNodeName,
+      String workflowAdjacencies) {
+    this(id, jobName, userName, submitTime, jobConfPath, jobACLs, queue,
+        workflowId, workflowName, workflowNodeName, workflowAdjacencies, "");
+  }
+
   /**
    * Create an event to record job submission
    * @param id The job Id of the job
@@ -83,12 +100,13 @@ public class JobSubmittedEvent implements HistoryEvent {
    * @param workflowName the workflow name
    * @param workflowNodeName the workflow node name
    * @param workflowAdjacencies the workflow adjacencies
+   * @param workflowTags Comma-separated workflow tags
    */
   public JobSubmittedEvent(JobID id, String jobName, String userName,
       long submitTime, String jobConfPath,
       Map<JobACL, AccessControlList> jobACLs, String queue,
       String workflowId, String workflowName, String workflowNodeName,
-      String workflowAdjacencies) {
+      String workflowAdjacencies, String workflowTags) {
     this.jobId = id;
     this.jobName = jobName;
     this.userName = userName;
@@ -100,6 +118,7 @@ public class JobSubmittedEvent implements HistoryEvent {
     this.workflowName = workflowName;
     this.workflowNodeName = workflowNodeName;
     this.workflowAdjacencies = workflowAdjacencies;
+    this.workflowTags = workflowTags;
   }
 
   /** Get the Job Id */
@@ -136,6 +155,10 @@ public class JobSubmittedEvent implements HistoryEvent {
   public String getWorkflowAdjacencies() {
     return workflowAdjacencies;
   }
+  /** Get the workflow tags */
+  public String getWorkflowTags() {
+    return workflowTags;
+  }
 
   /** Get the event type */
   public EventType getEventType() { return EventType.JOB_SUBMITTED; }