[BEAM-11712] Run TPC-DS via BeamSQL and Spark runner

sdks/java/testing/tpcds/README.md

@@ -0,0 +1,64 @@
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+-->
+
+# TPC-DS Benchmark
+
+## Google Dataflow Runner
+
+To execute TPC-DS benchmark for 1Gb dataset on Google Dataflow, run the following example command from the command line:
+
+```bash
+/gradlew :sdks:java:testing:tpcds:run -Ptpcds.args="--dataSize=1G \
+  --queries=3,26,55 \
+  --tpcParallel=2 \
+  --project=apache-beam-testing \
+  --stagingLocation=gs://beamsql_tpcds_1/staging \
+  --tempLocation=gs://beamsql_tpcds_2/temp \
+  --runner=DataflowRunner \
+  --region=us-west1 \
+  --maxNumWorkers=10"
+```
+
+To run a query using ZetaSQL planner (currently Query96 can be run using ZetaSQL), set the plannerName as below. If not specified, the default planner is Calcite.
+
+```bash
+./gradlew :sdks:java:testing:tpcds:run -Ptpcds.args="--dataSize=1G \
+  --queries=96 \
+  --tpcParallel=2 \
+  --plannerName=org.apache.beam.sdk.extensions.sql.zetasql.ZetaSQLQueryPlanner \
+  --project=apache-beam-testing \
+  --stagingLocation=gs://beamsql_tpcds_1/staging \
+  --tempLocation=gs://beamsql_tpcds_2/temp \
+  --runner=DataflowRunner \
+  --region=us-west1 \
+  --maxNumWorkers=10"
+```
+
+## Spark Runner
+
+To execute TPC-DS benchmark with Query3 for 1Gb dataset on Apache Spark 2.x, run the following example command from the command line:
+
+```bash
+./gradlew :sdks:java:testing:tpcds:run -Ptpcds.runner=":runners:spark:2" -Ptpcds.args=" \
+  --dataSize=1G \
+  --tpcParallel=1 \
+  --runner=SparkRunner --queries=3 \
+  --dataDirectory=/path/to/tpcds_data/ \
+  --resultsDirectory=/path/to/tpcds_results/"
+```

sdks/java/testing/tpcds/build.gradle

@@ -16,38 +16,106 @@
  * limitations under the License.
  */
 
-plugins {
-    id 'java'
-}
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(
+        automaticModuleName: 'org.apache.beam.sdk.tpcds',
+        exportJavadoc: false,
+        archivesBaseName: 'beam-sdks-java-tpcds',
+)
 
-description = "Apache Beam :: SDKs :: Java :: TPC-DS Benchark"
+description = "Apache Beam :: SDKs :: Java :: TPC-DS"
 
-version '2.24.0-SNAPSHOT'
+// When running via Gradle, this property can be used to pass commandline arguments
+// to the TPD-DS run
+def tpcdsArgsProperty = "tpcds.args"
 
-sourceCompatibility = 1.8
+// When running via Gradle, this property sets the runner dependency
+def tpcdsRunnerProperty = "tpcds.runner"
+def tpcdsRunnerDependency = project.findProperty(tpcdsRunnerProperty)
+        ?: ":runners:direct-java"
+def shouldProvideSpark = ":runners:spark:2".equals(tpcdsRunnerDependency)
+def isDataflowRunner = ":runners:google-cloud-dataflow-java".equals(tpcdsRunnerDependency)
+def runnerConfiguration = ":runners:direct-java".equals(tpcdsRunnerDependency) ? "shadow" : null
+
+if (isDataflowRunner) {
+    /*
+     * We need to rely on manually specifying these evaluationDependsOn to ensure that
+     * the following projects are evaluated before we evaluate this project. This is because
+     * we are attempting to reference a property from the project directly.
+     */
+    evaluationDependsOn(":runners:google-cloud-dataflow-java:worker:legacy-worker")
+}
 
-repositories {
-    mavenCentral()
+configurations {
+    // A configuration for running the TPC-DS launcher directly from Gradle, which
+    // uses Gradle to put the appropriate dependencies on the Classpath rather than
+    // bundling them into a fat jar
+    gradleRun
 }
 
 dependencies {
-    compile 'com.googlecode.json-simple:json-simple:1.1.1'
-    compile project(path: ":sdks:java:core", configuration: "shadow")
-    compile project(path: ":runners:google-cloud-dataflow-java")
-    compile project(":sdks:java:io:google-cloud-platform")
+    compile library.java.vendored_guava_26_0_jre
+    compile library.java.vendored_calcite_1_20_0
+    compile library.java.commons_csv
+    compile library.java.slf4j_api
+    compile "com.googlecode.json-simple:json-simple:1.1.1"
+    compile "com.alibaba:fastjson:1.2.69"
     compile project(":sdks:java:extensions:sql")
     compile project(":sdks:java:extensions:sql:zetasql")
-    compile group: 'com.google.auto.service', name: 'auto-service', version: '1.0-rc1'
-    testCompile group: 'junit', name: 'junit', version: '4.12'
+    compile project(path: ":runners:google-cloud-dataflow-java")
+    compile project(path: ":sdks:java:core", configuration: "shadow")
+    testRuntimeClasspath library.java.slf4j_jdk14
+    testCompile project(path: ":sdks:java:io:google-cloud-platform", configuration: "testRuntime")
+    gradleRun project(project.path)
+    gradleRun project(path: tpcdsRunnerDependency, configuration: runnerConfiguration)
+
+    // The Spark runner requires the user to provide a Spark dependency. For self-contained
+    // runs with the Spark runner, we can provide such a dependency. This is deliberately phrased
+    // to not hardcode any runner other than :runners:direct-java
+    if (shouldProvideSpark) {
+        gradleRun library.java.spark_core, {
+            exclude group:"org.slf4j", module:"jul-to-slf4j"
+        }
+        gradleRun library.java.spark_sql
+        gradleRun library.java.spark_streaming
+    }
 }
 
-// When running via Gradle, this property can be used to pass commandline arguments
-// to the tpcds run
-def tpcdsArgsProperty = "tpcds.args"
+if (shouldProvideSpark) {
+    configurations.gradleRun {
+        // Using Spark runner causes a StackOverflowError if slf4j-jdk14 is on the classpath
+        exclude group: "org.slf4j", module: "slf4j-jdk14"
+    }
+}
 
+// Execute the TPC-DS queries or suites via Gradle.
+//
+// Parameters:
+//   -Ptpcds.runner
+//       Specify a runner subproject, such as ":runners:spark:2" or ":runners:flink:1.10"
+//       Defaults to ":runners:direct-java"
+//
+//   -Ptpcds.args
+//       Specify the command line for invoking org.apache.beam.sdk.tpcds.BeamTpcds
 task run(type: JavaExec) {
-    main = "org.apache.beam.sdk.tpcds.BeamTpcds"
-    classpath = sourceSets.main.runtimeClasspath
     def tpcdsArgsStr = project.findProperty(tpcdsArgsProperty) ?: ""
-    args tpcdsArgsStr.split()
+    def tpcdsArgsList = new ArrayList<String>()
+    Collections.addAll(tpcdsArgsList, tpcdsArgsStr.split())
+
+    if (isDataflowRunner) {
+        dependsOn ":runners:google-cloud-dataflow-java:worker:legacy-worker:shadowJar"
+
+        def dataflowWorkerJar = project.findProperty('dataflowWorkerJar') ?:
+                project(":runners:google-cloud-dataflow-java:worker:legacy-worker")
+                        .shadowJar.archivePath
+        // Provide job with a customizable worker jar.
+        // With legacy worker jar, containerImage is set to empty (i.e. to use the internal build).
+        // More context and discussions can be found in PR#6694.
+        tpcdsArgsList.add("--dataflowWorkerJar=${dataflowWorkerJar}".toString())
+        tpcdsArgsList.add('--workerHarnessContainerImage=')
+    }
+
+    main = "org.apache.beam.sdk.tpcds.BeamTpcds"
+    classpath = configurations.gradleRun
+    args tpcdsArgsList.toArray()
 }