Merge branch 'rakeshmalladi-addWormholePushJobType'

Ragesh Rajagopalan · Ragesh Rajagopalan · commit 7f794bed44ee · 2018-09-12T09:14:02.000+05:30
diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md
@@ -155,3 +155,8 @@ The following were contributed by Chris Li.
 The following were contributed by Jamie Sun. Thanks, Jamie! 
 * `Add conditional workflow feature`
 * `Fix base properties scope for subflows when generating YAML files`
+* `Fix variable substitution issue when generating YAML files`
+* `Fix variable substitution issue for dependency list`
+
+The following were contributed by Rakesh Malladi. Thanks, Rakesh!
+* `Add ability for Hadoop DSL to understand the WormholePushJob job type`
diff --git a/VERSIONS.md b/VERSIONS.md
@@ -17,6 +17,15 @@ the License.
 Note that the LinkedIn build system occasionally requires that we skip a
 version bump, so you will see a few skipped version numbers in the list below.
 
+0.14.30
+* Adding WormholePushJob job type support
+
+0.14.29
+* Fix variable substitution issue for dependency list
+
+0.14.28
+* Fix variable substitution issue when generating YAML files
+
 0.14.27
 * Fix base properties scope for subflows when generating YAML files
 
diff --git a/gradle.properties b/gradle.properties
@@ -1,2 +1,2 @@
 org.gradle.daemon=true
-version=0.14.27
+version=0.14.30
diff --git a/hadoop-plugin-test/expectedJobs/flowWithVariableSubstitution/hadoop-plugin-test.project b/hadoop-plugin-test/expectedJobs/flowWithVariableSubstitution/hadoop-plugin-test.project
@@ -0,0 +1 @@
+azkaban-flow-version: 2.0
diff --git a/hadoop-plugin-test/expectedJobs/flowWithVariableSubstitution/wordCountFlow.flow b/hadoop-plugin-test/expectedJobs/flowWithVariableSubstitution/wordCountFlow.flow
@@ -0,0 +1,30 @@
+config:
+  user.to.proxy: foo
+nodes:
+- name: wordCountFlow
+  type: noop
+  dependsOn:
+  - checkResultsJob
+- name: checkResultsJob
+  type: hadoopShell
+  dependsOn:
+  - wordCountMapReduceJob
+  config:
+    command: bash ./bash/checkResults.sh . Government
+- name: wordCountMapReduceJob
+  type: hadoopJava
+  dependsOn:
+  - uploadResourceFilesJob
+  condition: ${uploadResourceFilesJob:param1} == "foo"
+  config:
+    job.class: com.linkedin.hadoop.example.WordCountJob
+    classpath: ./lib/*
+    input.path: ./text
+    output.path: ./word-count-map-reduce
+    hadoop-inject.mapreduce.job.classloader: 'true'
+    force.output.overwrite: 'true'
+- name: uploadResourceFilesJob
+  type: hadoopShell
+  config:
+    command: hadoop fs -mkdir -p .
+    command.1: hadoop fs -copyFromLocal -f ./text .
diff --git a/hadoop-plugin-test/expectedJobs/jobs1/jobs1.job b/hadoop-plugin-test/expectedJobs/jobs1/jobs1.job
@@ -1,3 +1,3 @@
 # This file generated from the Hadoop DSL. Do not edit by hand.
 type=noop
-dependencies=jobs1_job25
+dependencies=jobs1_job26
diff --git a/hadoop-plugin-test/expectedJobs/jobs1/jobs1_job26.job b/hadoop-plugin-test/expectedJobs/jobs1/jobs1_job26.job
@@ -0,0 +1,10 @@
+# This file generated from the Hadoop DSL. Do not edit by hand.
+type=WormholePushJob
+dependencies=jobs1_job25
+dataset=model
+dataset.group=linkedin-project
+input.path=/user/input
+namespace=linkedin
+output.location=LOCAL
+preserve.input=true
+version.string=1.2.4-2018.09.04
diff --git a/hadoop-plugin-test/expectedOutput/negative/missingFields.out b/hadoop-plugin-test/expectedOutput/negative/missingFields.out
@@ -44,5 +44,6 @@ RequiredFieldsChecker ERROR: PinotBuildAndPushJob job18 must set tableName, inpu
 RequiredFieldsChecker ERROR: TableauJob job19 must set workbookName
 RequiredFieldsChecker ERROR: HdfsWaitJob job20 must set dirPath, freshness, timeout, and forceJobToFail
 RequiredFieldsChecker ERROR: VenicePushJob job23 must set avroKeyField, avroValueField, clusterName, inputPath, veniceStoreName
+RequiredFieldsChecker ERROR: WormholePushJob job26 must set inputPath, namespace, datasetGroup and dataset
 RequiredFieldsChecker WARNING: Properties properties1 does not set any confProperties, jobProperties, jvmProperties or basePropertySetName. Nothing will be built for this properties object.
 :hadoop-plugin-test:test_missingFields FAILED
diff --git a/hadoop-plugin-test/expectedOutput/positive/flowWithVariableSubstitution.out b/hadoop-plugin-test/expectedOutput/positive/flowWithVariableSubstitution.out
@@ -0,0 +1,4 @@
+:hadoop-plugin-test:test_flowWithVariableSubstitution
+Running test for the file positive/flowWithVariableSubstitution.gradle
+Hadoop DSL static checker PASSED
+
diff --git a/hadoop-plugin-test/src/main/gradle/negative/missingFields.gradle b/hadoop-plugin-test/src/main/gradle/negative/missingFields.gradle
@@ -122,6 +122,10 @@ hadoop {
     tensorFlowJob('job25', 'spark') {
       depends 'job24'
     }
-    targets 'job25'
+
+    wormholePushJob('job26') {
+      depends 'job25'
+    }
+    targets 'job26'
   }
 }
diff --git a/hadoop-plugin-test/src/main/gradle/positive/flowWithVariableSubstitution.gradle b/hadoop-plugin-test/src/main/gradle/positive/flowWithVariableSubstitution.gradle
@@ -0,0 +1,69 @@
+buildscript {
+  repositories {
+    jcenter()
+  }
+  dependencies {
+    classpath files("${project.pluginTestDir}/hadoop-plugin-${project.version}.jar", "${project.pluginTestDir}/hadoop-plugin-${project.version}-SNAPSHOT.jar")
+    // Must manually specify snakeyaml for testing, not required for users
+    classpath 'org.yaml:snakeyaml:1.18'
+  }
+}
+
+apply plugin: com.linkedin.gradle.hadoop.HadoopPlugin
+
+// Test generating a flow YAML file with variable substitution "${}" in the job properties, job name, dependencies and conditions
+
+def projectPath = "."
+def jobName = "uploadResourceFilesJob"
+def jobName2 = "checkResultsJob"
+def dependList = ["${jobName}"]
+def targetList = ["${jobName2}"]
+
+hadoop {
+  buildPath "jobs/flowWithVariableSubstitution"
+  cleanPath true
+  generateYamlOutput true
+
+  propertyFile('common') {
+    set properties: [
+      'user.to.proxy': 'foo'
+    ]
+  }
+
+  workflow('wordCountFlow') {
+    hadoopShellJob("${jobName}") {
+      usesCommands([
+          "hadoop fs -mkdir -p ${projectPath}",
+          "hadoop fs -copyFromLocal -f ./text ${projectPath}"
+      ])
+    }
+
+    hadoopJavaJob('wordCountMapReduceJob') {
+      uses 'com.linkedin.hadoop.example.WordCountJob'
+      jvmClasspath './lib/*'
+      reads files: [
+          'input.path': "${projectPath}/text"
+      ]
+      writes files: [
+          'output.path': "${projectPath}/word-count-map-reduce"
+      ]
+      set confProperties: [
+          'mapreduce.job.classloader': true
+      ]
+      set properties: [
+          'force.output.overwrite': true
+      ]
+      depends dependList
+      conditions '${' + "${jobName}" + ':param1} == "foo"'
+    }
+
+    def keywordToCheck = "Government"
+
+    hadoopShellJob("${jobName2}") {
+      uses "bash ./bash/checkResults.sh ${projectPath} ${keywordToCheck}"
+      depends 'wordCountMapReduceJob'
+    }
+
+    targets targetList
+  }
+}
diff --git a/hadoop-plugin-test/src/main/gradle/positive/jobs1.gradle b/hadoop-plugin-test/src/main/gradle/positive/jobs1.gradle
@@ -419,6 +419,18 @@ hadoop {
       ]
       depends 'job24'
     }
-    targets 'job25'
+
+    wormholePushJob('job26') {
+      fromInputPath '/user/input'            // Required
+      toNamespace 'linkedin'                 // Required
+      toDatasetGroup 'linkedin-project'      // Required
+      toDataset 'model'                      // Required
+      withVersionString '1.2.4-2018.09.04'   // Optional
+      preserveInput true                     // Optional
+      toOutputLocation 'LOCAL'               // Optional
+
+      depends 'job25'
+    }
+    targets 'job26'
   }
 }
diff --git a/hadoop-plugin/src/main/groovy/com/linkedin/gradle/azkaban/AzkabanDslYamlCompiler.groovy b/hadoop-plugin/src/main/groovy/com/linkedin/gradle/azkaban/AzkabanDslYamlCompiler.groovy
@@ -296,7 +296,7 @@ class AzkabanDslYamlCompiler extends BaseCompiler {
     yamlizedJob["type"] = job.jobProperties["type"];
     // Add job dependencies if there are any
     if (!job.dependencyNames.isEmpty()) {
-      yamlizedJob["dependsOn"] = job.dependencyNames.toList();
+      yamlizedJob["dependsOn"] = job.dependencyNames.toList().collect{ dep -> dep.toString() };
     }
     if(job.condition != null) {
       yamlizedJob["condition"] = job.condition;
@@ -307,6 +307,11 @@ class AzkabanDslYamlCompiler extends BaseCompiler {
     config.remove("type");
     config.remove("dependencies");
     if (!config.isEmpty()) {
+      // If the job config contains "${}", which is a GString, it can involve lazy evaluation.
+      // It means it's not until the toString() method is invoked that the GString is evaluated.
+      // Without toString(), it can cause incompatible issue when writing the GString directly
+      // to YAML file.
+      config.each { key, val -> config[key] = val.toString()};
       yamlizedJob["config"] = config;
     }
 
diff --git a/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/BaseNamedScopeContainer.groovy b/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/BaseNamedScopeContainer.groovy
@@ -39,7 +39,7 @@ import com.linkedin.gradle.hadoopdsl.job.TensorFlowTonyJob;
 import com.linkedin.gradle.hadoopdsl.job.TeradataToHdfsJob;
 import com.linkedin.gradle.hadoopdsl.job.VenicePushJob;
 import com.linkedin.gradle.hadoopdsl.job.VoldemortBuildPushJob;
-
+import com.linkedin.gradle.hadoopdsl.job.WormholePushJob;
 import org.gradle.api.Project;
 
 /**
@@ -1116,6 +1116,19 @@ abstract class BaseNamedScopeContainer implements NamedScopeContainer {
     }
   }
 
+  /**
+   * DSL wormholePushJob method. Creates a wormholePushJob in scope with the given name
+   * and configuration.
+   *
+   * @param name The job name
+   * @param configure The configuration closure
+   * @return The new job
+   */
+  @HadoopDslMethod
+  WormholePushJob wormholePushJob(String name, @DelegatesTo(WormholePushJob) Closure configure) {
+    return ((WormholePushJob)configureJob(factory.makeWormholePushJob(name), configure));
+  }
+
   /**
    * DSL trigger method. Creates a Trigger in scope with the given name and configuration.
    *
diff --git a/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/BaseVisitor.groovy b/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/BaseVisitor.groovy
@@ -37,6 +37,7 @@ import com.linkedin.gradle.hadoopdsl.job.TensorFlowJob;
 import com.linkedin.gradle.hadoopdsl.job.TeradataToHdfsJob;
 import com.linkedin.gradle.hadoopdsl.job.VenicePushJob;
 import com.linkedin.gradle.hadoopdsl.job.VoldemortBuildPushJob;
+import com.linkedin.gradle.hadoopdsl.job.WormholePushJob;
 
 /**
  * Base implementation of Visitor to make it easy to implement visitor subclasses. By default, the
@@ -231,6 +232,11 @@ abstract class BaseVisitor implements Visitor {
     visitJob((Job)job);
   }
 
+  @Override
+  void visitJob(WormholePushJob job) {
+    visitJob((Job)job);
+  }
+
   @Override
   void visitJob(TensorFlowJob job) {
     visitJob((Job)job);
diff --git a/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/HadoopDslFactory.groovy b/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/HadoopDslFactory.groovy
@@ -42,6 +42,7 @@ import com.linkedin.gradle.hadoopdsl.job.TensorFlowTonyJob;
 import com.linkedin.gradle.hadoopdsl.job.TeradataToHdfsJob;
 import com.linkedin.gradle.hadoopdsl.job.VenicePushJob;
 import com.linkedin.gradle.hadoopdsl.job.VoldemortBuildPushJob;
+import com.linkedin.gradle.hadoopdsl.job.WormholePushJob;
 import com.linkedin.gradle.test.AssertionWorkflowExtension;
 import com.linkedin.gradle.test.TestExtension;
 
@@ -380,6 +381,16 @@ class HadoopDslFactory {
     return new Workflow(name, project, parentScope);
   }
 
+  /**
+   * Factory method to build a WormholePushJob.
+   *
+   * @param name The job name
+   * @return The job
+   */
+  WormholePushJob makeWormholePushJob(String name) {
+    return new WormholePushJob(name);
+  }
+
   /**
    * Factory method to build a Trigger.
    *
diff --git a/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/HadoopDslPlugin.groovy b/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/HadoopDslPlugin.groovy
@@ -134,6 +134,7 @@ class HadoopDslPlugin extends BaseNamedScopeContainer implements Plugin<Project>
     project.extensions.add("teradataToHdfsJob", this.&teradataToHdfsJob);
     project.extensions.add("venicePushJob", this.&venicePushJob);
     project.extensions.add("voldemortBuildPushJob", this.&voldemortBuildPushJob);
+    project.extensions.add("wormholePushJob", this.&wormholePushJob);
     project.extensions.add("tensorFlowJob", this.&tensorFlowJob);
   }
 
diff --git a/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/NamedScopeContainer.groovy b/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/NamedScopeContainer.groovy
@@ -37,6 +37,7 @@ import com.linkedin.gradle.hadoopdsl.job.TeradataToHdfsJob;
 import com.linkedin.gradle.hadoopdsl.job.TensorFlowJob;
 import com.linkedin.gradle.hadoopdsl.job.VenicePushJob;
 import com.linkedin.gradle.hadoopdsl.job.VoldemortBuildPushJob;
+import com.linkedin.gradle.hadoopdsl.job.WormholePushJob;
 
 /**
  * Interface that enables us to declare which DSL classes can declare a new scope (i.e. that have a
@@ -507,6 +508,16 @@ interface NamedScopeContainer {
   @HadoopDslMethod
   VenicePushJob venicePushJob(String name, @DelegatesTo(VenicePushJob) Closure configure);
 
+  /**
+   * DSL wormholePushJob method. Creates a WormholePushJob in scope with the given name and configuration.
+   *
+   * @param name The job name
+   * @param configure The configuration closure
+   * @return The new job
+   */
+  @HadoopDslMethod
+  WormholePushJob wormholePushJob(String name, @DelegatesTo(WormholePushJob) Closure configure);
+
   /**
    * DSL tensorFlowJob method. Creates a TensorFlowJob in scope with the given name and configuration.
    *
diff --git a/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/Visitor.groovy b/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/Visitor.groovy
@@ -37,6 +37,7 @@ import com.linkedin.gradle.hadoopdsl.job.TensorFlowJob;
 import com.linkedin.gradle.hadoopdsl.job.TeradataToHdfsJob;
 import com.linkedin.gradle.hadoopdsl.job.VenicePushJob;
 import com.linkedin.gradle.hadoopdsl.job.VoldemortBuildPushJob;
+import com.linkedin.gradle.hadoopdsl.job.WormholePushJob;
 
 /**
  * Defines a visitor pattern interface for the Hadoop DSL.
@@ -74,4 +75,5 @@ interface Visitor {
   void visitJob(TeradataToHdfsJob job);
   void visitJob(VenicePushJob job);
   void visitJob(VoldemortBuildPushJob job);
+  void visitJob(WormholePushJob job);
 }
diff --git a/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/checker/RequiredFieldsChecker.groovy b/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/checker/RequiredFieldsChecker.groovy
@@ -33,12 +33,11 @@ import com.linkedin.gradle.hadoopdsl.job.PigJob;
 import com.linkedin.gradle.hadoopdsl.job.PinotBuildAndPushJob;
 import com.linkedin.gradle.hadoopdsl.job.SparkJob;
 import com.linkedin.gradle.hadoopdsl.job.SqlJob;
-import com.linkedin.gradle.hadoopdsl.job.TableauJob;
-import com.linkedin.gradle.hadoopdsl.job.TensorFlowJob;
+import com.linkedin.gradle.hadoopdsl.job.TableauJob
 import com.linkedin.gradle.hadoopdsl.job.TeradataToHdfsJob;
 import com.linkedin.gradle.hadoopdsl.job.VenicePushJob;
 import com.linkedin.gradle.hadoopdsl.job.VoldemortBuildPushJob;
-
+import com.linkedin.gradle.hadoopdsl.job.WormholePushJob;
 import org.gradle.api.Project;
 
 /**
@@ -232,6 +231,19 @@ class RequiredFieldsChecker extends BaseStaticChecker {
     }
   }
 
+  @Override
+  void visitJob(WormholePushJob job) {
+    boolean emptyInputPath = job.inputPath == null || job.inputPath.isEmpty();
+    boolean emptyNamespace = job.namespace == null || job.namespace.isEmpty();
+    boolean emptyDatasetGroup = job.datasetGroup == null || job.datasetGroup.isEmpty();
+    boolean emptyDataset = job.dataset == null || job.dataset.isEmpty();
+
+    if (emptyInputPath || emptyNamespace || emptyDatasetGroup || emptyDataset) {
+      project.logger.lifecycle("RequiredFieldsChecker ERROR: WormholePushJob ${job.name} must set inputPath, namespace, datasetGroup and dataset");
+      foundError = true;
+    }
+  }
+
   @Override
   void visitJob(HdfsToTeradataJob job) {
     foundError |= validateNotEmpty(job, "hostName", job.hostName);
diff --git a/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/job/WormholePushJob.groovy b/hadoop-plugin/src/main/groovy/com/linkedin/gradle/hadoopdsl/job/WormholePushJob.groovy
diff --git a/li-hadoop-plugin/src/main/groovy/com/linkedin/gradle/liazkaban/LiAzkabanDslYamlCompiler.groovy b/li-hadoop-plugin/src/main/groovy/com/linkedin/gradle/liazkaban/LiAzkabanDslYamlCompiler.groovy

Original file line number	Diff line number	Diff line change
`@@ -1,2 +1,2 @@`
`1`	`1`	`org.gradle.daemon=true`
`2`		`-version=0.14.27`
	`2`	`+version=0.14.30`
Original file line number	Diff line number	Diff line change
`@@ -122,6 +122,10 @@ hadoop {`
`122`	`122`	`tensorFlowJob('job25', 'spark') {`
`123`	`123`	`depends 'job24'`
`124`	`124`	`}`
`125`		`- targets 'job25'`
	`125`	`+`
	`126`	`+ wormholePushJob('job26') {`
	`127`	`+ depends 'job25'`
	`128`	`+ }`
	`129`	`+ targets 'job26'`
`126`	`130`	`}`
`127`	`131`	`}`