You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/04/20 01:35:55 UTC
[GitHub] [spark] maropu opened a new pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
maropu opened a new pull request #32243:
URL: https://github.com/apache/spark/pull/32243
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
This PR proposes to port minimal code to generate TPC-DS data from [databricks/spark-sql-perf](https://github.com/databricks/spark-sql-perf). The classes in a new class file `tpcdsDatagen.scala` are basically copied from the `databricks/spark-sql-perf` codebase.
Note that I've modified them a bit to follow the Spark code style and removed unnecessary parts from them.
The code authors of these classes are:
@juliuszsompolski
@npoggi
@wangyum
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
We frequently use TPCDS data now for benchmarks/tests, but the classes for the TPCDS schemas of datagen and benchmarks/tests are managed separately, e.g.,
- https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/TPCDSBase.scala
- https://github.com/databricks/spark-sql-perf/blob/master/src/main/scala/com/databricks/spark/sql/perf/tpcds/TPCDSTables.scala
I think this causes some inconveniences, e.g., we need to update both files in the separate repositories if we update the TPCDS schema #32037. So, it would be useful for the Spark codebase to generate them by referring to the same schema definition.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
dev only.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
-->
Manually checked and GA passed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830254801
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138110/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616337927
##########
File path: .github/workflows/build_and_test.yml
##########
@@ -528,6 +515,24 @@ jobs:
uses: actions/setup-java@v1
with:
java-version: 8
+ - name: Cache TPC-DS generated data
+ id: cache-tpcds-sf-1
+ uses: actions/cache@v2
+ with:
+ path: ./tpcds-sf-1
+ key: tpcds-${{ hashFiles('sql/core/src/test/scala/org/apache/spark/sql/TPCDSSchema.scala') }}
Review comment:
Is this sufficient?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-829756832
**[Test build #138095 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138095/testReport)** for PR 32243 at commit [`ef68b61`](https://github.com/apache/spark/commit/ef68b61a8d5cad890a010ec3b02459dd396c6000).
* This patch **fails RAT tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yaooqinn commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616288243
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q23a.sql.out
##########
@@ -3,4 +3,4 @@
-- !query schema
struct<sum(sales):decimal(28,2)>
-- !query output
-17030.91
+NULL
Review comment:
Does the NULL look right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830129849
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42631/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r624230151
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/tpcdsDagagen.scala
##########
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.immutable.Stream
+import scala.sys.process._
+import scala.util.Try
+
+import org.apache.spark.SparkContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.functions.{col, rpad}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
+
+// The classes in this file are basically moved from https://github.com/databricks/spark-sql-perf
+
+/**
+ * Using ProcessBuilder.lineStream produces a stream, that uses
+ * a LinkedBlockingQueue with a default capacity of Integer.MAX_VALUE.
+ *
+ * This causes OOM if the consumer cannot keep up with the producer.
+ *
+ * See scala.sys.process.ProcessBuilderImpl.lineStream
+ */
+object BlockingLineStream {
+
+ // See scala.sys.process.Streamed
+ private final class BlockingStreamed[T](
+ val process: T => Unit,
+ val done: Int => Unit,
+ val stream: () => Stream[T])
+
+ // See scala.sys.process.Streamed
+ private object BlockingStreamed {
+ // scala.process.sys.Streamed uses default of Integer.MAX_VALUE,
+ // which causes OOMs if the consumer cannot keep up with producer.
+ val maxQueueSize = 65536
+
+ def apply[T](nonzeroException: Boolean): BlockingStreamed[T] = {
+ val q = new LinkedBlockingQueue[Either[Int, T]](maxQueueSize)
+
+ def next(): Stream[T] = q.take match {
+ case Left(0) => Stream.empty
+ case Left(code) =>
+ if (nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
+ case Right(s) => Stream.cons(s, next())
+ }
+
+ new BlockingStreamed((s: T) => q put Right(s), code => q put Left(code), () => next())
+ }
+ }
+
+ // See scala.sys.process.ProcessImpl.Spawn
+ private object Spawn {
+ def apply(f: => Unit): Thread = apply(f, daemon = false)
+ def apply(f: => Unit, daemon: Boolean): Thread = {
+ val thread = new Thread() { override def run() = { f } }
+ thread.setDaemon(daemon)
+ thread.start()
+ thread
+ }
+ }
+
+ def apply(command: Seq[String]): Stream[String] = {
+ val streamed = BlockingStreamed[String](true)
+ val process = command.run(BasicIO(false, streamed.process, None))
+ Spawn(streamed.done(process.exitValue()))
+ streamed.stream()
+ }
+}
+
+class Dsdgen(dsdgenDir: String) extends Serializable {
+ private val dsdgen = s"$dsdgenDir/dsdgen"
+
+ def generate(
+ sparkContext: SparkContext,
+ tableName: String,
+ partitions: Int,
+ scaleFactor: Int): RDD[String] = {
+ val generatedData = {
+ sparkContext.parallelize(1 to partitions, partitions).flatMap { i =>
+ val localToolsDir = if (new java.io.File(dsdgen).exists) {
+ dsdgenDir
+ } else if (new java.io.File(s"/$dsdgen").exists) {
+ s"/$dsdgenDir"
+ } else {
+ throw new IllegalStateException(
+ s"Could not find dsdgen at $dsdgen or /$dsdgen. Run install")
+ }
+
+ // NOTE: RNGSEED is the RNG seed used by the data generator. Right now, it is fixed to
+ // 19620718 that is used to generate `https://github.com/maropu/spark-tpcds-sf-1`.
+ val parallel = if (partitions > 1) s"-parallel $partitions -child $i" else ""
+ val commands = Seq(
+ "bash", "-c",
+ s"cd $localToolsDir && ./dsdgen -table $tableName -filter Y -scale $scaleFactor " +
+ s"-RNGSEED 19620718 $parallel")
+ BlockingLineStream(commands)
+ }
+ }
+
+ generatedData.setName(s"$tableName, sf=$scaleFactor, strings")
+ generatedData
+ }
+}
+
+class TPCDSTables(sqlContext: SQLContext, dsdgenDir: String, scaleFactor: Int)
+ extends TPCDSSchema with Logging with Serializable {
+
+ private val dataGenerator = new Dsdgen(dsdgenDir)
+
+ private def tables: Seq[Table] = tableColumns.map { case (tableName, schemaString) =>
+ val partitionColumns = tablePartitionColumns.getOrElse(tableName, Nil)
+ .map(_.stripPrefix("`").stripSuffix("`"))
+ Table(tableName, partitionColumns, StructType.fromDDL(schemaString))
+ }.toSeq
+
+ private case class Table(name: String, partitionColumns: Seq[String], schema: StructType) {
+ def nonPartitioned: Table = {
+ Table(name, Nil, schema)
+ }
+
+ private def df(numPartition: Int) = {
+ val generatedData = dataGenerator.generate(
+ sqlContext.sparkContext, name, numPartition, scaleFactor)
+ val rows = generatedData.mapPartitions { iter =>
+ iter.map { l =>
+ val values = l.split("\\|", -1).dropRight(1).map { v =>
+ if (v.equals("")) {
+ // If the string value is an empty string, we turn it to a null
+ null
+ } else {
+ v
+ }
+ }
+ Row.fromSeq(values)
+ }
+ }
+
+ val stringData =
+ sqlContext.createDataFrame(
+ rows,
+ StructType(schema.fields.map(f => StructField(f.name, StringType))))
+
+ val convertedData = {
+ val columns = schema.fields.map { f =>
+ val c = f.dataType match {
+ // Needs right-padding for char types
+ case CharType(n) => rpad(Column(f.name), n, " ")
+ // Don't need a cast for varchar types
+ case _: VarcharType => col(f.name)
+ case _ => col(f.name).cast(f.dataType)
+ }
+ c.as(f.name)
+ }
+ stringData.select(columns: _*)
+ }
+
+ convertedData
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ numPartitions: Int): Unit = {
+ val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Ignore
+
+ val data = df(numPartitions)
+ val tempTableName = s"${name}_text"
+ data.createOrReplaceTempView(tempTableName)
+
+ val writer = if (partitionColumns.nonEmpty) {
+ if (clusterByPartitionColumns) {
+ val columnString = data.schema.fields.map { field =>
+ field.name
+ }.mkString(",")
+ val partitionColumnString = partitionColumns.mkString(",")
+ val predicates = if (filterOutNullPartitionValues) {
+ partitionColumns.map(col => s"$col IS NOT NULL").mkString("WHERE ", " AND ", "")
+ } else {
+ ""
+ }
+
+ val query =
+ s"""
+ |SELECT
+ | $columnString
+ |FROM
+ | $tempTableName
+ |$predicates
+ |DISTRIBUTE BY
+ | $partitionColumnString
+ """.stripMargin
+ val grouped = sqlContext.sql(query)
+ logInfo(s"Pre-clustering with partitioning columns with query $query.")
+ grouped.write
+ } else {
+ data.write
+ }
+ } else {
+ // treat non-partitioned tables as "one partition" that we want to coalesce
+ if (clusterByPartitionColumns) {
+ // in case data has more than maxRecordsPerFile, split into multiple writers to improve
+ // datagen speed files will be truncated to maxRecordsPerFile value, so the final
+ // result will be the same.
+ val numRows = data.count
+ val maxRecordPerFile = Try {
+ sqlContext.getConf("spark.sql.files.maxRecordsPerFile").toInt
+ }.getOrElse(0)
+
+ if (maxRecordPerFile > 0 && numRows > maxRecordPerFile) {
+ val numFiles = (numRows.toDouble/maxRecordPerFile).ceil.toInt
+ logInfo(s"Coalescing into $numFiles files")
+ data.coalesce(numFiles).write
+ } else {
+ data.coalesce(1).write
+ }
+ } else {
+ data.write
+ }
+ }
+ writer.format(format).mode(mode)
+ if (partitionColumns.nonEmpty) {
+ writer.partitionBy(partitionColumns: _*)
+ }
+ logInfo(s"Generating table $name in database to $location with save mode $mode.")
+ writer.save(location)
+ sqlContext.dropTempTable(tempTableName)
+ }
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ partitionTables: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ tableFilter: String = "",
+ numPartitions: Int = 100): Unit = {
+ var tablesToBeGenerated = if (partitionTables) {
+ tables
+ } else {
+ tables.map(_.nonPartitioned)
+ }
+
+ if (!tableFilter.isEmpty) {
+ tablesToBeGenerated = tablesToBeGenerated.filter(_.name == tableFilter)
+ if (tablesToBeGenerated.isEmpty) {
+ throw new RuntimeException("Bad table name filter: " + tableFilter)
+ }
+ }
+
+ tablesToBeGenerated.foreach { table =>
+ val tableLocation = s"$location/${table.name}"
+ table.genData(tableLocation, format, overwrite, clusterByPartitionColumns,
+ filterOutNullPartitionValues, numPartitions)
+ }
+ }
+}
+
+class GenTPCDSDataConfig(args: Array[String]) {
+ var master: String = "local[*]"
+ var dsdgenDir: String = null
+ var location: String = null
+ var scaleFactor: Int = 1
+ var format: String = "parquet"
+ var overwrite: Boolean = false
+ var partitionTables: Boolean = false
+ var clusterByPartitionColumns: Boolean = false
+ var filterOutNullPartitionValues: Boolean = false
+ var tableFilter: String = ""
+ var numPartitions: Int = 100
+
+ parseArgs(args.toList)
+
+ private def parseArgs(inputArgs: List[String]): Unit = {
+ var args = inputArgs
+
+ while (args.nonEmpty) {
+ args match {
+ case "--master" :: value :: tail =>
+ master = value
+ args = tail
+
+ case "--dsdgenDir" :: value :: tail =>
+ dsdgenDir = value
+ args = tail
+
+ case "--location" :: value :: tail =>
+ location = value
+ args = tail
+
+ case "--scaleFactor" :: value :: tail =>
+ scaleFactor = toPositiveIntValue("Scale factor", value)
+ args = tail
+
+ case "--format" :: value :: tail =>
+ format = value
+ args = tail
+
+ case "--overwrite" :: tail =>
+ overwrite = true
+ args = tail
+
+ case "--partitionTables" :: tail =>
+ partitionTables = true
+ args = tail
+
+ case "--clusterByPartitionColumns" :: tail =>
+ clusterByPartitionColumns = true
+ args = tail
+
+ case "--filterOutNullPartitionValues" :: tail =>
+ filterOutNullPartitionValues = true
+ args = tail
+
+ case "--tableFilter" :: value :: tail =>
+ tableFilter = value
+ args = tail
+
+ case "--numPartitions" :: value :: tail =>
+ numPartitions = toPositiveIntValue("Number of partitions", value)
+ args = tail
+
+ case "--help" :: tail =>
+ printUsageAndExit(0)
+
+ case _ =>
+ // scalastyle:off println
+ System.err.println("Unknown/unsupported param " + args)
+ // scalastyle:on println
+ printUsageAndExit(1)
+ }
+ }
+
+ checkRequiredArguments()
+ }
+
+ private def printUsageAndExit(exitCode: Int): Unit = {
+ // scalastyle:off
+ System.err.println("""
+ |build/sbt "test:runMain <this class> [Options]"
+ |Options:
+ | --master the Spark master to use, default to local[*]
+ | --dsdgenDir location of dsdgen
+ | --location root directory of location to generate data in
+ | --scaleFactor size of the dataset to generate (in GB)
+ | --format generated data format, Parquet, ORC ...
+ | --overwrite whether to overwrite the data that is already there
+ | --partitionTables whether to create the partitioned fact tables
+ | --clusterByPartitionColumns whether to shuffle to get partitions coalesced into single files
+ | --filterOutNullPartitionValues whether to filter out the partition with NULL key value
+ | --tableFilter comma-separated list of table names to generate (e.g., store_sales,store_returns),
+ | all the tables are generated by default
+ | --numPartitions how many dsdgen partitions to run - number of input tasks
+ """.stripMargin)
+ // scalastyle:on
+ System.exit(exitCode)
+ }
+
+ private def toPositiveIntValue(name: String, v: String): Int = {
+ if (Try(v.toInt).getOrElse(-1) <= 0) {
+ // scalastyle:off println
+ System.err.println(s"$name must be a positive number")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ v.toInt
+ }
+
+ private def checkRequiredArguments(): Unit = {
+ if (dsdgenDir == null) {
+ // scalastyle:off println
+ System.err.println("Must specify a dsdgen path")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ if (location == null) {
+ // scalastyle:off println
+ System.err.println("Must specify an output location")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ }
+}
+
+/**
+ * This class generates TPCDS table data by using tpcds-kit:
+ * - https://github.com/databricks/tpcds-kit
+ *
+ * To run this:
+ * {{{
+ * build/sbt "sql/test:runMain <this class> --dsdgenDir <path> --location <path> --scaleFactor 1"
+ * }}}
+ */
+object GenTPCDSData {
Review comment:
sure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822928442
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42187/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822914568
**[Test build #137658 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137658/testReport)** for PR 32243 at commit [`7b6949d`](https://github.com/apache/spark/commit/7b6949ddb3b2a99856c24dd6d6d0487281b530e1).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830536383
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138119/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616283865
##########
File path: .github/workflows/build_and_test.yml
##########
@@ -528,6 +515,24 @@ jobs:
uses: actions/setup-java@v1
with:
java-version: 8
+ - name: Cache TPC-DS generated data
+ id: cache-tpcds-sf-1
+ uses: actions/cache@v2
+ with:
+ path: ./tpcds-sf-1
+ key: tpcds-${{ hashFiles('sql/core/src/test/scala/org/apache/spark/sql/TPCDSSchema.scala') }}
+ - name: Checkout tpcds-kit repository
+ if: steps.cache-tpcds-sf-1.outputs.cache-hit != 'true'
+ uses: actions/checkout@v2
+ with:
+ repository: databricks/tpcds-kit
+ path: ./tpcds-kit
+ - name: Build tpcds-kit
+ if: steps.cache-tpcds-sf-1.outputs.cache-hit != 'true'
+ run: cd tpcds-kit/tools && make OS=LINUX
Review comment:
The part above was copied from https://github.com/apache/spark/pull/31303/files#diff-48c0ee97c53013d18d6bbae44648f7fab9af2e0bf5b0dc1ca761e18ec5c478f2 cc: @wangyum
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yaooqinn commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616288734
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q24b.sql.out
##########
@@ -3,4 +3,4 @@
-- !query schema
struct<c_last_name:string,c_first_name:string,s_store_name:string,paid:decimal(27,2)>
-- !query output
-Griffith Ray able 161564.48
Review comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-829756844
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138095/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822936436
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42187/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830473391
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42640/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830439073
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42638/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830539675
okay, ready to review. cc: @juliuszsompolski @npoggi @wangyum
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822973963
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137658/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu closed pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu closed pull request #32243:
URL: https://github.com/apache/spark/pull/32243
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yaooqinn commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616288827
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q24a.sql.out
##########
@@ -3,10 +3,4 @@
-- !query schema
struct<c_last_name:string,c_first_name:string,s_store_name:string,paid:decimal(27,2)>
-- !query output
-Holt Curtis able 8225.80
-Kunz Lee able 34631.52
-Littlefield Clarence able 127380.00
-NULL Tommy able 38118.08
-Pettit Richard able 3930.52
-Townsend Franklin able 68983.20
-Winchester Margaret bar 14269.20
+
Review comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830417418
**[Test build #138117 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138117/testReport)** for PR 32243 at commit [`a88fe32`](https://github.com/apache/spark/commit/a88fe32b241ec1f1a2674b69944640e1445d71cf).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830217846
Thank you for update, @maropu . The updated version looks neat!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830516100
**[Test build #138119 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138119/testReport)** for PR 32243 at commit [`5f4d022`](https://github.com/apache/spark/commit/5f4d022f93e7da391fb0b58131b0191c8fd65edc).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830475565
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42640/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830417418
**[Test build #138117 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138117/testReport)** for PR 32243 at commit [`a88fe32`](https://github.com/apache/spark/commit/a88fe32b241ec1f1a2674b69944640e1445d71cf).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616362212
##########
File path: .github/workflows/build_and_test.yml
##########
@@ -528,6 +515,24 @@ jobs:
uses: actions/setup-java@v1
with:
java-version: 8
+ - name: Cache TPC-DS generated data
+ id: cache-tpcds-sf-1
+ uses: actions/cache@v2
+ with:
+ path: ./tpcds-sf-1
+ key: tpcds-${{ hashFiles('sql/core/src/test/scala/org/apache/spark/sql/TPCDSSchema.scala') }}
Review comment:
yes, I think it is for now since this PR targets at schema definition changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822914568
**[Test build #137658 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137658/testReport)** for PR 32243 at commit [`7b6949d`](https://github.com/apache/spark/commit/7b6949ddb3b2a99856c24dd6d6d0487281b530e1).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yaooqinn commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822916118
Great work, @maropu and thanks for making this happen
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830254801
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138110/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-829782931
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42615/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616338684
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q1.sql.out
##########
@@ -3,103 +3,103 @@
-- !query schema
struct<c_customer_id:string>
-- !query output
-AAAAAAAAAAABBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAAKAAAA
-AAAAAAAAAABDAAAA
-AAAAAAAAAABHBAAA
-AAAAAAAAAABLAAAA
-AAAAAAAAAABMAAAA
-AAAAAAAAAACHAAAA
-AAAAAAAAAACMAAAA
-AAAAAAAAAADDAAAA
-AAAAAAAAAADGAAAA
-AAAAAAAAAADGBAAA
-AAAAAAAAAADGBAAA
+AAAAAAAAAAABAAAA
Review comment:
In this case, it seems that we had better split this PR into two. WDYT, @maropu ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-824462833
I think it's okay. As you pointed out, GitHub Actions already caches the generated SF1 data one, and it doesn't seem adding much codes for that.
> we need to update spark-sql-perf to the latest TPC-DS spec and query templates
@maropu would you mind porting the TPC-DS spec and query update to `spark-sql-perf` when you find some time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830253956
**[Test build #138110 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138110/testReport)** for PR 32243 at commit [`5ee9c1a`](https://github.com/apache/spark/commit/5ee9c1a0f87233ed1b188dddcead9199c466c5c5).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830462044
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138117/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r624011869
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/tpcdsDagagen.scala
##########
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.immutable.Stream
+import scala.sys.process._
+import scala.util.Try
+
+import org.apache.spark.SparkContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.functions.{col, rpad}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
+
+// The classes in this file are basically moved from https://github.com/databricks/spark-sql-perf
+
+/**
+ * Using ProcessBuilder.lineStream produces a stream, that uses
+ * a LinkedBlockingQueue with a default capacity of Integer.MAX_VALUE.
+ *
+ * This causes OOM if the consumer cannot keep up with the producer.
+ *
+ * See scala.sys.process.ProcessBuilderImpl.lineStream
+ */
+object BlockingLineStream {
+
+ // See scala.sys.process.Streamed
+ private final class BlockingStreamed[T](
+ val process: T => Unit,
+ val done: Int => Unit,
+ val stream: () => Stream[T])
+
+ // See scala.sys.process.Streamed
+ private object BlockingStreamed {
+ // scala.process.sys.Streamed uses default of Integer.MAX_VALUE,
+ // which causes OOMs if the consumer cannot keep up with producer.
+ val maxQueueSize = 65536
+
+ def apply[T](nonzeroException: Boolean): BlockingStreamed[T] = {
+ val q = new LinkedBlockingQueue[Either[Int, T]](maxQueueSize)
+
+ def next(): Stream[T] = q.take match {
+ case Left(0) => Stream.empty
+ case Left(code) =>
+ if (nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
+ case Right(s) => Stream.cons(s, next())
+ }
+
+ new BlockingStreamed((s: T) => q put Right(s), code => q put Left(code), () => next())
+ }
+ }
+
+ // See scala.sys.process.ProcessImpl.Spawn
+ private object Spawn {
+ def apply(f: => Unit): Thread = apply(f, daemon = false)
+ def apply(f: => Unit, daemon: Boolean): Thread = {
+ val thread = new Thread() { override def run() = { f } }
+ thread.setDaemon(daemon)
+ thread.start()
+ thread
+ }
+ }
+
+ def apply(command: Seq[String]): Stream[String] = {
+ val streamed = BlockingStreamed[String](true)
+ val process = command.run(BasicIO(false, streamed.process, None))
+ Spawn(streamed.done(process.exitValue()))
+ streamed.stream()
+ }
+}
+
+class Dsdgen(dsdgenDir: String) extends Serializable {
+ private val dsdgen = s"$dsdgenDir/dsdgen"
+
+ def generate(
+ sparkContext: SparkContext,
+ tableName: String,
+ partitions: Int,
+ scaleFactor: Int): RDD[String] = {
+ val generatedData = {
+ sparkContext.parallelize(1 to partitions, partitions).flatMap { i =>
+ val localToolsDir = if (new java.io.File(dsdgen).exists) {
+ dsdgenDir
+ } else if (new java.io.File(s"/$dsdgen").exists) {
+ s"/$dsdgenDir"
+ } else {
+ throw new IllegalStateException(
+ s"Could not find dsdgen at $dsdgen or /$dsdgen. Run install")
+ }
+
+ // NOTE: RNGSEED is the RNG seed used by the data generator. Right now, it is fixed to
+ // 19620718 that is used to generate `https://github.com/maropu/spark-tpcds-sf-1`.
+ val parallel = if (partitions > 1) s"-parallel $partitions -child $i" else ""
+ val commands = Seq(
+ "bash", "-c",
+ s"cd $localToolsDir && ./dsdgen -table $tableName -filter Y -scale $scaleFactor " +
+ s"-RNGSEED 19620718 $parallel")
+ BlockingLineStream(commands)
+ }
+ }
+
+ generatedData.setName(s"$tableName, sf=$scaleFactor, strings")
+ generatedData
+ }
+}
+
+class TPCDSTables(sqlContext: SQLContext, dsdgenDir: String, scaleFactor: Int)
+ extends TPCDSSchema with Logging with Serializable {
+
+ private val dataGenerator = new Dsdgen(dsdgenDir)
+
+ private def tables: Seq[Table] = tableColumns.map { case (tableName, schemaString) =>
+ val partitionColumns = tablePartitionColumns.getOrElse(tableName, Nil)
+ .map(_.stripPrefix("`").stripSuffix("`"))
+ Table(tableName, partitionColumns, StructType.fromDDL(schemaString))
+ }.toSeq
+
+ private case class Table(name: String, partitionColumns: Seq[String], schema: StructType) {
+ def nonPartitioned: Table = {
+ Table(name, Nil, schema)
+ }
+
+ private def df(numPartition: Int) = {
+ val generatedData = dataGenerator.generate(
+ sqlContext.sparkContext, name, numPartition, scaleFactor)
+ val rows = generatedData.mapPartitions { iter =>
+ iter.map { l =>
+ val values = l.split("\\|", -1).dropRight(1).map { v =>
+ if (v.equals("")) {
+ // If the string value is an empty string, we turn it to a null
+ null
+ } else {
+ v
+ }
+ }
+ Row.fromSeq(values)
+ }
+ }
+
+ val stringData =
+ sqlContext.createDataFrame(
+ rows,
+ StructType(schema.fields.map(f => StructField(f.name, StringType))))
+
+ val convertedData = {
+ val columns = schema.fields.map { f =>
+ val c = f.dataType match {
+ // Needs right-padding for char types
+ case CharType(n) => rpad(Column(f.name), n, " ")
+ // Don't need a cast for varchar types
+ case _: VarcharType => col(f.name)
+ case _ => col(f.name).cast(f.dataType)
+ }
+ c.as(f.name)
+ }
+ stringData.select(columns: _*)
+ }
+
+ convertedData
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ numPartitions: Int): Unit = {
+ val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Ignore
+
+ val data = df(numPartitions)
+ val tempTableName = s"${name}_text"
+ data.createOrReplaceTempView(tempTableName)
+
+ val writer = if (partitionColumns.nonEmpty) {
+ if (clusterByPartitionColumns) {
+ val columnString = data.schema.fields.map { field =>
+ field.name
+ }.mkString(",")
+ val partitionColumnString = partitionColumns.mkString(",")
+ val predicates = if (filterOutNullPartitionValues) {
+ partitionColumns.map(col => s"$col IS NOT NULL").mkString("WHERE ", " AND ", "")
+ } else {
+ ""
+ }
+
+ val query =
+ s"""
+ |SELECT
+ | $columnString
+ |FROM
+ | $tempTableName
+ |$predicates
+ |DISTRIBUTE BY
+ | $partitionColumnString
+ """.stripMargin
+ val grouped = sqlContext.sql(query)
+ logInfo(s"Pre-clustering with partitioning columns with query $query.")
+ grouped.write
+ } else {
+ data.write
+ }
+ } else {
+ // treat non-partitioned tables as "one partition" that we want to coalesce
+ if (clusterByPartitionColumns) {
+ // in case data has more than maxRecordsPerFile, split into multiple writers to improve
+ // datagen speed files will be truncated to maxRecordsPerFile value, so the final
+ // result will be the same.
+ val numRows = data.count
+ val maxRecordPerFile = Try {
+ sqlContext.getConf("spark.sql.files.maxRecordsPerFile").toInt
+ }.getOrElse(0)
+
+ if (maxRecordPerFile > 0 && numRows > maxRecordPerFile) {
+ val numFiles = (numRows.toDouble/maxRecordPerFile).ceil.toInt
+ logInfo(s"Coalescing into $numFiles files")
+ data.coalesce(numFiles).write
+ } else {
+ data.coalesce(1).write
+ }
+ } else {
+ data.write
+ }
+ }
+ writer.format(format).mode(mode)
+ if (partitionColumns.nonEmpty) {
+ writer.partitionBy(partitionColumns: _*)
+ }
+ logInfo(s"Generating table $name in database to $location with save mode $mode.")
+ writer.save(location)
+ sqlContext.dropTempTable(tempTableName)
+ }
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ partitionTables: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ tableFilter: String = "",
+ numPartitions: Int = 100): Unit = {
+ var tablesToBeGenerated = if (partitionTables) {
+ tables
+ } else {
+ tables.map(_.nonPartitioned)
+ }
+
+ if (!tableFilter.isEmpty) {
+ tablesToBeGenerated = tablesToBeGenerated.filter(_.name == tableFilter)
+ if (tablesToBeGenerated.isEmpty) {
+ throw new RuntimeException("Bad table name filter: " + tableFilter)
+ }
+ }
+
+ tablesToBeGenerated.foreach { table =>
+ val tableLocation = s"$location/${table.name}"
+ table.genData(tableLocation, format, overwrite, clusterByPartitionColumns,
+ filterOutNullPartitionValues, numPartitions)
+ }
+ }
+}
+
+class GenTPCDSDataConfig(args: Array[String]) {
+ var master: String = "local[*]"
+ var dsdgenDir: String = null
+ var location: String = null
+ var scaleFactor: Int = 1
+ var format: String = "parquet"
+ var overwrite: Boolean = false
+ var partitionTables: Boolean = false
+ var clusterByPartitionColumns: Boolean = false
+ var filterOutNullPartitionValues: Boolean = false
+ var tableFilter: String = ""
+ var numPartitions: Int = 100
+
+ parseArgs(args.toList)
+
+ private def parseArgs(inputArgs: List[String]): Unit = {
+ var args = inputArgs
+
+ while (args.nonEmpty) {
+ args match {
+ case "--master" :: value :: tail =>
+ master = value
+ args = tail
+
+ case "--dsdgenDir" :: value :: tail =>
+ dsdgenDir = value
+ args = tail
+
+ case "--location" :: value :: tail =>
+ location = value
+ args = tail
+
+ case "--scaleFactor" :: value :: tail =>
+ scaleFactor = toPositiveIntValue("Scale factor", value)
+ args = tail
+
+ case "--format" :: value :: tail =>
+ format = value
+ args = tail
+
+ case "--overwrite" :: tail =>
+ overwrite = true
+ args = tail
+
+ case "--partitionTables" :: tail =>
+ partitionTables = true
+ args = tail
+
+ case "--clusterByPartitionColumns" :: tail =>
+ clusterByPartitionColumns = true
+ args = tail
+
+ case "--filterOutNullPartitionValues" :: tail =>
+ filterOutNullPartitionValues = true
+ args = tail
+
+ case "--tableFilter" :: value :: tail =>
+ tableFilter = value
+ args = tail
+
+ case "--numPartitions" :: value :: tail =>
+ numPartitions = toPositiveIntValue("Number of partitions", value)
+ args = tail
+
+ case "--help" :: tail =>
+ printUsageAndExit(0)
+
+ case _ =>
+ // scalastyle:off println
+ System.err.println("Unknown/unsupported param " + args)
+ // scalastyle:on println
+ printUsageAndExit(1)
+ }
+ }
+
+ checkRequiredArguments()
+ }
+
+ private def printUsageAndExit(exitCode: Int): Unit = {
+ // scalastyle:off
+ System.err.println("""
+ |build/sbt "test:runMain <this class> [Options]"
+ |Options:
+ | --master the Spark master to use, default to local[*]
+ | --dsdgenDir location of dsdgen
Review comment:
Is this the only difference from your repo, isn't it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830464661
**[Test build #138119 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138119/testReport)** for PR 32243 at commit [`5f4d022`](https://github.com/apache/spark/commit/5f4d022f93e7da391fb0b58131b0191c8fd65edc).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon edited a comment on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
HyukjinKwon edited a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-824462833
I think it's okay. As you pointed out, GitHub Actions already caches the generated SF1 data one, and it doesn't seem adding much codes for that.
> we need to update spark-sql-perf to the latest TPC-DS spec and query templates
@maropu would you mind porting the TPC-DS spec and query update to `spark-sql-perf` when you find some time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-829756844
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138095/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830141599
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42631/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-824827326
Ur..., I noticed that generated data are different between the GA env(Ubuntu) and my env(MacOS) by following[ the same workflow](https://github.com/apache/spark/pull/32243/files#diff-48c0ee97c53013d18d6bbae44648f7fab9af2e0bf5b0dc1ca761e18ec5c478f2R524-R535). Probably, the generator behaviour seems to depend on the implementation of random functions. I'm currently not sure that we can generate the same data between different linux distro, so I need more work (e.g., adding a script to generate data on docker env) for making it easy for developers to generate data/golden files....
> @maropu would you mind porting the TPC-DS spec and query update to spark-sql-perf when you find some time?
Of course not. Probably, I have time to work on it tomorrow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r623873687
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q1.sql.out
##########
@@ -3,103 +3,103 @@
-- !query schema
struct<c_customer_id:string>
-- !query output
-AAAAAAAAAAABBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAAKAAAA
-AAAAAAAAAABDAAAA
-AAAAAAAAAABHBAAA
-AAAAAAAAAABLAAAA
-AAAAAAAAAABMAAAA
-AAAAAAAAAACHAAAA
-AAAAAAAAAACMAAAA
-AAAAAAAAAADDAAAA
-AAAAAAAAAADGAAAA
-AAAAAAAAAADGBAAA
-AAAAAAAAAADGBAAA
+AAAAAAAAAAABAAAA
Review comment:
okay, I've fixed the code so that it generate the same TPC-DS data with https://github.com/maropu/spark-tpcds-sf-1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-829756672
**[Test build #138095 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138095/testReport)** for PR 32243 at commit [`ef68b61`](https://github.com/apache/spark/commit/ef68b61a8d5cad890a010ec3b02459dd396c6000).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616284573
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q1.sql.out
##########
@@ -3,103 +3,103 @@
-- !query schema
struct<c_customer_id:string>
-- !query output
-AAAAAAAAAAABBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAAKAAAA
-AAAAAAAAAABDAAAA
-AAAAAAAAAABHBAAA
-AAAAAAAAAABLAAAA
-AAAAAAAAAABMAAAA
-AAAAAAAAAACHAAAA
-AAAAAAAAAACMAAAA
-AAAAAAAAAADDAAAA
-AAAAAAAAAADGAAAA
-AAAAAAAAAADGBAAA
-AAAAAAAAAADGBAAA
+AAAAAAAAAAABAAAA
Review comment:
This PR updated the golden files because datagen used to generate `maropu/spark-tpcds-sf-1` was older than the `databricks/tpcds-kit` one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822936436
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42187/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830997133
Thank you, @HyukjinKwon @dongjoon-hyun ~ Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830536383
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138119/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] npoggi commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
npoggi commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-831482901
Arriving a big late. Looks good. We should move the DDL for the tables as resource files at some point. Thanks for the update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830464661
**[Test build #138119 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138119/testReport)** for PR 32243 at commit [`5f4d022`](https://github.com/apache/spark/commit/5f4d022f93e7da391fb0b58131b0191c8fd65edc).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r623872242
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/tpcdsDagagen.scala
##########
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.immutable.Stream
+import scala.sys.process._
+import scala.util.Try
+
+import org.apache.spark.SparkContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.functions.{col, rpad}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
+
+// The classes in this file are basically moved from https://github.com/databricks/spark-sql-perf
+
+/**
+ * Using ProcessBuilder.lineStream produces a stream, that uses
+ * a LinkedBlockingQueue with a default capacity of Integer.MAX_VALUE.
+ *
+ * This causes OOM if the consumer cannot keep up with the producer.
+ *
+ * See scala.sys.process.ProcessBuilderImpl.lineStream
+ */
+object BlockingLineStream {
+
+ // See scala.sys.process.Streamed
+ private final class BlockingStreamed[T](
+ val process: T => Unit,
+ val done: Int => Unit,
+ val stream: () => Stream[T])
+
+ // See scala.sys.process.Streamed
+ private object BlockingStreamed {
+ // scala.process.sys.Streamed uses default of Integer.MAX_VALUE,
+ // which causes OOMs if the consumer cannot keep up with producer.
+ val maxQueueSize = 65536
+
+ def apply[T](nonzeroException: Boolean): BlockingStreamed[T] = {
+ val q = new LinkedBlockingQueue[Either[Int, T]](maxQueueSize)
+
+ def next(): Stream[T] = q.take match {
+ case Left(0) => Stream.empty
+ case Left(code) =>
+ if (nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
+ case Right(s) => Stream.cons(s, next())
+ }
+
+ new BlockingStreamed((s: T) => q put Right(s), code => q put Left(code), () => next())
+ }
+ }
+
+ // See scala.sys.process.ProcessImpl.Spawn
+ private object Spawn {
+ def apply(f: => Unit): Thread = apply(f, daemon = false)
+ def apply(f: => Unit, daemon: Boolean): Thread = {
+ val thread = new Thread() { override def run() = { f } }
+ thread.setDaemon(daemon)
+ thread.start()
+ thread
+ }
+ }
+
+ def apply(command: Seq[String]): Stream[String] = {
+ val streamed = BlockingStreamed[String](true)
+ val process = command.run(BasicIO(false, streamed.process, None))
+ Spawn(streamed.done(process.exitValue()))
+ streamed.stream()
+ }
+}
+
+class Dsdgen(dsdgenDir: String) extends Serializable {
+ private val dsdgen = s"$dsdgenDir/dsdgen"
+
+ def generate(
+ sparkContext: SparkContext,
+ tableName: String,
+ partitions: Int,
+ scaleFactor: Int): RDD[String] = {
+ val generatedData = {
+ sparkContext.parallelize(1 to partitions, partitions).flatMap { i =>
+ val localToolsDir = if (new java.io.File(dsdgen).exists) {
+ dsdgenDir
+ } else if (new java.io.File(s"/$dsdgen").exists) {
+ s"/$dsdgenDir"
+ } else {
+ throw new IllegalStateException(
+ s"Could not find dsdgen at $dsdgen or /$dsdgen. Run install")
+ }
+
+ // NOTE: RNGSEED is the RNG seed used by the data generator. Right now, it is fixed to
+ // 19620718 that is used to generate `https://github.com/maropu/spark-tpcds-sf-1`.
+ val parallel = if (partitions > 1) s"-parallel $partitions -child $i" else ""
+ val commands = Seq(
+ "bash", "-c",
+ s"cd $localToolsDir && ./dsdgen -table $tableName -filter Y -scale $scaleFactor " +
+ s"-RNGSEED 19620718 $parallel")
Review comment:
NOTE: In a follow-up PR, I'll revert back this value to 100 to use `https://github.com/databricks/tpcds-kit` instead. See: https://github.com/apache/spark/pull/32243#issuecomment-824827326
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yaooqinn commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616288776
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q25.sql.out
##########
@@ -3,4 +3,4 @@
-- !query schema
struct<i_item_id:string,i_item_desc:string,s_store_id:string,s_store_name:string,store_sales_profit:decimal(17,2),store_returns_loss:decimal(17,2),catalog_sales_profit:decimal(17,2)>
-- !query output
-AAAAAAAADPMBAAAA Things know alone letters. Flights should tend even jewish fees. Civil plans could not cry also social days; other losses might not pay walls; still able signs should not remove too human AAAAAAAAHAAAAAAA ation 12.84 91.41 -1329.46
Review comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830439073
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42638/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616359685
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q1.sql.out
##########
@@ -3,103 +3,103 @@
-- !query schema
struct<c_customer_id:string>
-- !query output
-AAAAAAAAAAABBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAAKAAAA
-AAAAAAAAAABDAAAA
-AAAAAAAAAABHBAAA
-AAAAAAAAAABLAAAA
-AAAAAAAAAABMAAAA
-AAAAAAAAAACHAAAA
-AAAAAAAAAACMAAAA
-AAAAAAAAAADDAAAA
-AAAAAAAAAADGAAAA
-AAAAAAAAAADGBAAA
-AAAAAAAAAADGBAAA
+AAAAAAAAAAABAAAA
Review comment:
I'll do it later, too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616289179
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q23b.sql.out
##########
@@ -3,7 +3,4 @@
-- !query schema
struct<c_last_name:string,c_first_name:string,sales:decimal(28,2)>
-- !query output
-NULL Robert 598.86
-Brown Monika 6031.52
-Collins Gordon 727.57
-Green Jesse 9672.96
+
Review comment:
Yea, I need to check the answers query-by-query later...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830141599
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42631/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822967198
**[Test build #137658 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137658/testReport)** for PR 32243 at commit [`7b6949d`](https://github.com/apache/spark/commit/7b6949ddb3b2a99856c24dd6d6d0487281b530e1).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `trait TPCDSBase extends SharedSparkSession with TPCDSSchema `
* `trait TPCDSSchema `
* `class Dsdgen(dsdgenDir: String) extends Serializable `
* `class TPCDSTables(sqlContext: SQLContext, dsdgenDir: String, scaleFactor: Int)`
* `class GenTPCDSDataConfig(args: Array[String]) `
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu edited a comment on pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu edited a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-824827326
~Ur..., I noticed that generated data are different between the GA env(Ubuntu) and my env(MacOS) by following[ the same workflow](https://github.com/apache/spark/pull/32243/files#diff-48c0ee97c53013d18d6bbae44648f7fab9af2e0bf5b0dc1ca761e18ec5c478f2R524-R535). Probably, the generator behaviour seems to depend on the implementation of random functions. I'm currently not sure that we can generate the same data between different linux distro, so I need more work (e.g., adding a script to generate data on docker env) for making it easy for developers to generate data/golden files....~
My bad. I misunderstood it and I just used different seeds `RNGSEED` when generating data. I've checked that the current code can generate the same TPC-DS data between different env (macos, Ubuntu, Amazon Linux 2, ...).
> @maropu would you mind porting the TPC-DS spec and query update to spark-sql-perf when you find some time?
Of course not. Probably, I have time to work on it tomorrow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r624013166
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/tpcdsDagagen.scala
##########
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.immutable.Stream
+import scala.sys.process._
+import scala.util.Try
+
+import org.apache.spark.SparkContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.functions.{col, rpad}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
+
+// The classes in this file are basically moved from https://github.com/databricks/spark-sql-perf
+
+/**
+ * Using ProcessBuilder.lineStream produces a stream, that uses
+ * a LinkedBlockingQueue with a default capacity of Integer.MAX_VALUE.
+ *
+ * This causes OOM if the consumer cannot keep up with the producer.
+ *
+ * See scala.sys.process.ProcessBuilderImpl.lineStream
+ */
+object BlockingLineStream {
+
+ // See scala.sys.process.Streamed
+ private final class BlockingStreamed[T](
+ val process: T => Unit,
+ val done: Int => Unit,
+ val stream: () => Stream[T])
+
+ // See scala.sys.process.Streamed
+ private object BlockingStreamed {
+ // scala.process.sys.Streamed uses default of Integer.MAX_VALUE,
+ // which causes OOMs if the consumer cannot keep up with producer.
+ val maxQueueSize = 65536
+
+ def apply[T](nonzeroException: Boolean): BlockingStreamed[T] = {
+ val q = new LinkedBlockingQueue[Either[Int, T]](maxQueueSize)
+
+ def next(): Stream[T] = q.take match {
+ case Left(0) => Stream.empty
+ case Left(code) =>
+ if (nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
+ case Right(s) => Stream.cons(s, next())
+ }
+
+ new BlockingStreamed((s: T) => q put Right(s), code => q put Left(code), () => next())
+ }
+ }
+
+ // See scala.sys.process.ProcessImpl.Spawn
+ private object Spawn {
+ def apply(f: => Unit): Thread = apply(f, daemon = false)
+ def apply(f: => Unit, daemon: Boolean): Thread = {
+ val thread = new Thread() { override def run() = { f } }
+ thread.setDaemon(daemon)
+ thread.start()
+ thread
+ }
+ }
+
+ def apply(command: Seq[String]): Stream[String] = {
+ val streamed = BlockingStreamed[String](true)
+ val process = command.run(BasicIO(false, streamed.process, None))
+ Spawn(streamed.done(process.exitValue()))
+ streamed.stream()
+ }
+}
+
+class Dsdgen(dsdgenDir: String) extends Serializable {
+ private val dsdgen = s"$dsdgenDir/dsdgen"
+
+ def generate(
+ sparkContext: SparkContext,
+ tableName: String,
+ partitions: Int,
+ scaleFactor: Int): RDD[String] = {
+ val generatedData = {
+ sparkContext.parallelize(1 to partitions, partitions).flatMap { i =>
+ val localToolsDir = if (new java.io.File(dsdgen).exists) {
+ dsdgenDir
+ } else if (new java.io.File(s"/$dsdgen").exists) {
+ s"/$dsdgenDir"
+ } else {
+ throw new IllegalStateException(
+ s"Could not find dsdgen at $dsdgen or /$dsdgen. Run install")
+ }
+
+ // NOTE: RNGSEED is the RNG seed used by the data generator. Right now, it is fixed to
+ // 19620718 that is used to generate `https://github.com/maropu/spark-tpcds-sf-1`.
+ val parallel = if (partitions > 1) s"-parallel $partitions -child $i" else ""
+ val commands = Seq(
+ "bash", "-c",
+ s"cd $localToolsDir && ./dsdgen -table $tableName -filter Y -scale $scaleFactor " +
+ s"-RNGSEED 19620718 $parallel")
+ BlockingLineStream(commands)
+ }
+ }
+
+ generatedData.setName(s"$tableName, sf=$scaleFactor, strings")
+ generatedData
+ }
+}
+
+class TPCDSTables(sqlContext: SQLContext, dsdgenDir: String, scaleFactor: Int)
+ extends TPCDSSchema with Logging with Serializable {
+
+ private val dataGenerator = new Dsdgen(dsdgenDir)
+
+ private def tables: Seq[Table] = tableColumns.map { case (tableName, schemaString) =>
+ val partitionColumns = tablePartitionColumns.getOrElse(tableName, Nil)
+ .map(_.stripPrefix("`").stripSuffix("`"))
+ Table(tableName, partitionColumns, StructType.fromDDL(schemaString))
+ }.toSeq
+
+ private case class Table(name: String, partitionColumns: Seq[String], schema: StructType) {
+ def nonPartitioned: Table = {
+ Table(name, Nil, schema)
+ }
+
+ private def df(numPartition: Int) = {
+ val generatedData = dataGenerator.generate(
+ sqlContext.sparkContext, name, numPartition, scaleFactor)
+ val rows = generatedData.mapPartitions { iter =>
+ iter.map { l =>
+ val values = l.split("\\|", -1).dropRight(1).map { v =>
+ if (v.equals("")) {
+ // If the string value is an empty string, we turn it to a null
+ null
+ } else {
+ v
+ }
+ }
+ Row.fromSeq(values)
+ }
+ }
+
+ val stringData =
+ sqlContext.createDataFrame(
+ rows,
+ StructType(schema.fields.map(f => StructField(f.name, StringType))))
+
+ val convertedData = {
+ val columns = schema.fields.map { f =>
+ val c = f.dataType match {
+ // Needs right-padding for char types
+ case CharType(n) => rpad(Column(f.name), n, " ")
+ // Don't need a cast for varchar types
+ case _: VarcharType => col(f.name)
+ case _ => col(f.name).cast(f.dataType)
+ }
+ c.as(f.name)
+ }
+ stringData.select(columns: _*)
+ }
+
+ convertedData
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ numPartitions: Int): Unit = {
+ val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Ignore
+
+ val data = df(numPartitions)
+ val tempTableName = s"${name}_text"
+ data.createOrReplaceTempView(tempTableName)
+
+ val writer = if (partitionColumns.nonEmpty) {
+ if (clusterByPartitionColumns) {
+ val columnString = data.schema.fields.map { field =>
+ field.name
+ }.mkString(",")
+ val partitionColumnString = partitionColumns.mkString(",")
+ val predicates = if (filterOutNullPartitionValues) {
+ partitionColumns.map(col => s"$col IS NOT NULL").mkString("WHERE ", " AND ", "")
+ } else {
+ ""
+ }
+
+ val query =
+ s"""
+ |SELECT
+ | $columnString
+ |FROM
+ | $tempTableName
+ |$predicates
+ |DISTRIBUTE BY
+ | $partitionColumnString
+ """.stripMargin
+ val grouped = sqlContext.sql(query)
+ logInfo(s"Pre-clustering with partitioning columns with query $query.")
+ grouped.write
+ } else {
+ data.write
+ }
+ } else {
+ // treat non-partitioned tables as "one partition" that we want to coalesce
+ if (clusterByPartitionColumns) {
+ // in case data has more than maxRecordsPerFile, split into multiple writers to improve
+ // datagen speed files will be truncated to maxRecordsPerFile value, so the final
+ // result will be the same.
+ val numRows = data.count
+ val maxRecordPerFile = Try {
+ sqlContext.getConf("spark.sql.files.maxRecordsPerFile").toInt
+ }.getOrElse(0)
+
+ if (maxRecordPerFile > 0 && numRows > maxRecordPerFile) {
+ val numFiles = (numRows.toDouble/maxRecordPerFile).ceil.toInt
+ logInfo(s"Coalescing into $numFiles files")
+ data.coalesce(numFiles).write
+ } else {
+ data.coalesce(1).write
+ }
+ } else {
+ data.write
+ }
+ }
+ writer.format(format).mode(mode)
+ if (partitionColumns.nonEmpty) {
+ writer.partitionBy(partitionColumns: _*)
+ }
+ logInfo(s"Generating table $name in database to $location with save mode $mode.")
+ writer.save(location)
+ sqlContext.dropTempTable(tempTableName)
+ }
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ partitionTables: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ tableFilter: String = "",
+ numPartitions: Int = 100): Unit = {
+ var tablesToBeGenerated = if (partitionTables) {
+ tables
+ } else {
+ tables.map(_.nonPartitioned)
+ }
+
+ if (!tableFilter.isEmpty) {
+ tablesToBeGenerated = tablesToBeGenerated.filter(_.name == tableFilter)
+ if (tablesToBeGenerated.isEmpty) {
+ throw new RuntimeException("Bad table name filter: " + tableFilter)
+ }
+ }
+
+ tablesToBeGenerated.foreach { table =>
+ val tableLocation = s"$location/${table.name}"
+ table.genData(tableLocation, format, overwrite, clusterByPartitionColumns,
+ filterOutNullPartitionValues, numPartitions)
+ }
+ }
+}
+
+class GenTPCDSDataConfig(args: Array[String]) {
+ var master: String = "local[*]"
+ var dsdgenDir: String = null
+ var location: String = null
+ var scaleFactor: Int = 1
+ var format: String = "parquet"
+ var overwrite: Boolean = false
+ var partitionTables: Boolean = false
+ var clusterByPartitionColumns: Boolean = false
+ var filterOutNullPartitionValues: Boolean = false
+ var tableFilter: String = ""
+ var numPartitions: Int = 100
+
+ parseArgs(args.toList)
+
+ private def parseArgs(inputArgs: List[String]): Unit = {
+ var args = inputArgs
+
+ while (args.nonEmpty) {
+ args match {
+ case "--master" :: value :: tail =>
+ master = value
+ args = tail
+
+ case "--dsdgenDir" :: value :: tail =>
+ dsdgenDir = value
+ args = tail
+
+ case "--location" :: value :: tail =>
+ location = value
+ args = tail
+
+ case "--scaleFactor" :: value :: tail =>
+ scaleFactor = toPositiveIntValue("Scale factor", value)
+ args = tail
+
+ case "--format" :: value :: tail =>
+ format = value
+ args = tail
+
+ case "--overwrite" :: tail =>
+ overwrite = true
+ args = tail
+
+ case "--partitionTables" :: tail =>
+ partitionTables = true
+ args = tail
+
+ case "--clusterByPartitionColumns" :: tail =>
+ clusterByPartitionColumns = true
+ args = tail
+
+ case "--filterOutNullPartitionValues" :: tail =>
+ filterOutNullPartitionValues = true
+ args = tail
+
+ case "--tableFilter" :: value :: tail =>
+ tableFilter = value
+ args = tail
+
+ case "--numPartitions" :: value :: tail =>
+ numPartitions = toPositiveIntValue("Number of partitions", value)
+ args = tail
+
+ case "--help" :: tail =>
+ printUsageAndExit(0)
+
+ case _ =>
+ // scalastyle:off println
+ System.err.println("Unknown/unsupported param " + args)
+ // scalastyle:on println
+ printUsageAndExit(1)
+ }
+ }
+
+ checkRequiredArguments()
+ }
+
+ private def printUsageAndExit(exitCode: Int): Unit = {
+ // scalastyle:off
+ System.err.println("""
+ |build/sbt "test:runMain <this class> [Options]"
+ |Options:
+ | --master the Spark master to use, default to local[*]
+ | --dsdgenDir location of dsdgen
+ | --location root directory of location to generate data in
+ | --scaleFactor size of the dataset to generate (in GB)
+ | --format generated data format, Parquet, ORC ...
+ | --overwrite whether to overwrite the data that is already there
+ | --partitionTables whether to create the partitioned fact tables
+ | --clusterByPartitionColumns whether to shuffle to get partitions coalesced into single files
+ | --filterOutNullPartitionValues whether to filter out the partition with NULL key value
+ | --tableFilter comma-separated list of table names to generate (e.g., store_sales,store_returns),
+ | all the tables are generated by default
+ | --numPartitions how many dsdgen partitions to run - number of input tasks
+ """.stripMargin)
+ // scalastyle:on
+ System.exit(exitCode)
+ }
+
+ private def toPositiveIntValue(name: String, v: String): Int = {
+ if (Try(v.toInt).getOrElse(-1) <= 0) {
+ // scalastyle:off println
+ System.err.println(s"$name must be a positive number")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ v.toInt
+ }
+
+ private def checkRequiredArguments(): Unit = {
+ if (dsdgenDir == null) {
+ // scalastyle:off println
+ System.err.println("Must specify a dsdgen path")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ if (location == null) {
+ // scalastyle:off println
+ System.err.println("Must specify an output location")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ }
+}
+
+/**
+ * This class generates TPCDS table data by using tpcds-kit:
+ * - https://github.com/databricks/tpcds-kit
+ *
+ * To run this:
+ * {{{
+ * build/sbt "sql/test:runMain <this class> --dsdgenDir <path> --location <path> --scaleFactor 1"
+ * }}}
+ */
+object GenTPCDSData {
+
+ def main(args: Array[String]): Unit = {
+ val config = new GenTPCDSDataConfig(args)
+
+ val spark = SparkSession
+ .builder()
+ .appName(getClass.getName)
+ .master(config.master)
+ .getOrCreate()
+
+ val tables = new TPCDSTables(
+ spark.sqlContext,
+ dsdgenDir = config.dsdgenDir,
+ scaleFactor = config.scaleFactor)
+
+ tables.genData(
+ location = config.location,
+ format = config.format,
+ overwrite = config.overwrite,
+ partitionTables = config.partitionTables,
+ clusterByPartitionColumns = config.clusterByPartitionColumns,
+ filterOutNullPartitionValues = config.filterOutNullPartitionValues,
+ tableFilter = config.tableFilter,
+ numPartitions = config.numPartitions)
Review comment:
We need `spark.stop()` after this line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616285491
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/tpcdsDagagen.scala
##########
@@ -0,0 +1,442 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.immutable.Stream
+import scala.sys.process._
+import scala.util.Try
+
+import org.apache.spark.SparkContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.functions.{col, rpad}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
+
+// The classes in this file are basically moved from https://github.com/databricks/spark-sql-perf
+
+/**
+ * Using ProcessBuilder.lineStream produces a stream, that uses
+ * a LinkedBlockingQueue with a default capacity of Integer.MAX_VALUE.
+ *
+ * This causes OOM if the consumer cannot keep up with the producer.
+ *
+ * See scala.sys.process.ProcessBuilderImpl.lineStream
+ */
+object BlockingLineStream {
+
+ // See scala.sys.process.Streamed
+ private final class BlockingStreamed[T](
+ val process: T => Unit,
+ val done: Int => Unit,
+ val stream: () => Stream[T])
+
+ // See scala.sys.process.Streamed
+ private object BlockingStreamed {
+ // scala.process.sys.Streamed uses default of Integer.MAX_VALUE,
+ // which causes OOMs if the consumer cannot keep up with producer.
+ val maxQueueSize = 65536
+
+ def apply[T](nonzeroException: Boolean): BlockingStreamed[T] = {
+ val q = new LinkedBlockingQueue[Either[Int, T]](maxQueueSize)
+
+ def next(): Stream[T] = q.take match {
+ case Left(0) => Stream.empty
+ case Left(code) =>
+ if (nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
+ case Right(s) => Stream.cons(s, next())
+ }
+
+ new BlockingStreamed((s: T) => q put Right(s), code => q put Left(code), () => next())
+ }
+ }
+
+ // See scala.sys.process.ProcessImpl.Spawn
+ private object Spawn {
+ def apply(f: => Unit): Thread = apply(f, daemon = false)
+ def apply(f: => Unit, daemon: Boolean): Thread = {
+ val thread = new Thread() { override def run() = { f } }
+ thread.setDaemon(daemon)
+ thread.start()
+ thread
+ }
+ }
+
+ def apply(command: Seq[String]): Stream[String] = {
+ val streamed = BlockingStreamed[String](true)
+ val process = command.run(BasicIO(false, streamed.process, None))
+ Spawn(streamed.done(process.exitValue()))
+ streamed.stream()
+ }
+}
+
+class Dsdgen(dsdgenDir: String) extends Serializable {
+ private val dsdgen = s"$dsdgenDir/dsdgen"
+
+ def generate(
+ sparkContext: SparkContext,
+ tableName: String,
+ partitions: Int,
+ scaleFactor: Int): RDD[String] = {
+ val generatedData = {
+ sparkContext.parallelize(1 to partitions, partitions).flatMap { i =>
+ val localToolsDir = if (new java.io.File(dsdgen).exists) {
+ dsdgenDir
+ } else if (new java.io.File(s"/$dsdgen").exists) {
+ s"/$dsdgenDir"
+ } else {
+ throw new IllegalStateException(
+ s"Could not find dsdgen at $dsdgen or /$dsdgen. Run install")
+ }
+
+ // NOTE: RNGSEED is the RNG seed used by the data generator. Right now, it is fixed to 100.
+ val parallel = if (partitions > 1) s"-parallel $partitions -child $i" else ""
+ val commands = Seq(
+ "bash", "-c",
+ s"cd $localToolsDir && ./dsdgen -table $tableName -filter Y -scale $scaleFactor " +
+ s"-RNGSEED 100 $parallel")
+ BlockingLineStream(commands)
+ }
+ }
+
+ generatedData.setName(s"$tableName, sf=$scaleFactor, strings")
+ generatedData
+ }
+}
+
+class TPCDSTables(sqlContext: SQLContext, dsdgenDir: String, scaleFactor: Int)
+ extends TPCDSSchema with Logging with Serializable {
+
+ private val dataGenerator = new Dsdgen(dsdgenDir)
+
+ private def tables: Seq[Table] = tableColumns.map { case (tableName, schemaString) =>
+ val partitionColumns = tablePartitionColumns.getOrElse(tableName, Nil)
+ .map(_.stripPrefix("`").stripSuffix("`"))
+ Table(tableName, partitionColumns, StructType.fromDDL(schemaString))
+ }.toSeq
+
+ private case class Table(name: String, partitionColumns: Seq[String], schema: StructType) {
+ def nonPartitioned: Table = {
+ Table(name, Nil, schema)
+ }
+
+ private def df(numPartition: Int) = {
+ val generatedData = dataGenerator.generate(
+ sqlContext.sparkContext, name, numPartition, scaleFactor)
+ val rows = generatedData.mapPartitions { iter =>
+ iter.map { l =>
+ val values = l.split("\\|", -1).dropRight(1).map { v =>
+ if (v.equals("")) {
+ // If the string value is an empty string, we turn it to a null
+ null
+ } else {
+ v
+ }
+ }
+ Row.fromSeq(values)
+ }
+ }
+
+ val stringData =
+ sqlContext.createDataFrame(
+ rows,
+ StructType(schema.fields.map(f => StructField(f.name, StringType))))
+
+ val convertedData = {
+ val columns = schema.fields.map { f =>
+ val c = f.dataType match {
+ // Needs right-padding for char types
+ case CharType(n) => rpad(Column(f.name), n, " ")
+ // Don't need a cast for varchar types
+ case _: VarcharType => col(f.name)
+ case _ => col(f.name).cast(f.dataType)
+ }
+ c.as(f.name)
+ }
+ stringData.select(columns: _*)
+ }
+
+ convertedData
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ numPartitions: Int): Unit = {
+ val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Ignore
+
+ val data = df(numPartitions)
+ val tempTableName = s"${name}_text"
+ data.createOrReplaceTempView(tempTableName)
+
+ val writer = if (partitionColumns.nonEmpty) {
+ if (clusterByPartitionColumns) {
+ val columnString = data.schema.fields.map { field =>
+ field.name
+ }.mkString(",")
+ val partitionColumnString = partitionColumns.mkString(",")
+ val predicates = if (filterOutNullPartitionValues) {
+ partitionColumns.map(col => s"$col IS NOT NULL").mkString("WHERE ", " AND ", "")
+ } else {
+ ""
+ }
+
+ val query =
+ s"""
+ |SELECT
+ | $columnString
+ |FROM
+ | $tempTableName
+ |$predicates
+ |DISTRIBUTE BY
+ | $partitionColumnString
+ """.stripMargin
+ val grouped = sqlContext.sql(query)
+ logInfo(s"Pre-clustering with partitioning columns with query $query.")
+ grouped.write
+ } else {
+ data.write
+ }
+ } else {
+ // treat non-partitioned tables as "one partition" that we want to coalesce
+ if (clusterByPartitionColumns) {
+ // in case data has more than maxRecordsPerFile, split into multiple writers to improve
+ // datagen speed files will be truncated to maxRecordsPerFile value, so the final
+ // result will be the same.
+ val numRows = data.count
+ val maxRecordPerFile = Try {
+ sqlContext.getConf("spark.sql.files.maxRecordsPerFile").toInt
+ }.getOrElse(0)
+
+ if (maxRecordPerFile > 0 && numRows > maxRecordPerFile) {
+ val numFiles = (numRows.toDouble/maxRecordPerFile).ceil.toInt
+ logInfo(s"Coalescing into $numFiles files")
+ data.coalesce(numFiles).write
+ } else {
+ data.coalesce(1).write
+ }
+ } else {
+ data.write
+ }
+ }
+ writer.format(format).mode(mode)
+ if (partitionColumns.nonEmpty) {
+ writer.partitionBy(partitionColumns: _*)
+ }
+ logInfo(s"Generating table $name in database to $location with save mode $mode.")
+ writer.save(location)
+ sqlContext.dropTempTable(tempTableName)
+ }
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ partitionTables: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ tableFilter: String = "",
+ numPartitions: Int = 100): Unit = {
+ var tablesToBeGenerated = if (partitionTables) {
+ tables
+ } else {
+ tables.map(_.nonPartitioned)
+ }
+
+ if (!tableFilter.isEmpty) {
+ tablesToBeGenerated = tablesToBeGenerated.filter(_.name == tableFilter)
+ if (tablesToBeGenerated.isEmpty) {
+ throw new RuntimeException("Bad table name filter: " + tableFilter)
+ }
+ }
+
+ tablesToBeGenerated.foreach { table =>
+ val tableLocation = s"$location/${table.name}"
+ table.genData(tableLocation, format, overwrite, clusterByPartitionColumns,
+ filterOutNullPartitionValues, numPartitions)
+ }
+ }
+}
+
+class GenTPCDSDataConfig(args: Array[String]) {
+ var master: String = "local[*]"
+ var dsdgenDir: String = null
+ var location: String = null
+ var scaleFactor: Int = 1
+ var format: String = "parquet"
+ var overwrite: Boolean = false
+ var partitionTables: Boolean = false
+ var clusterByPartitionColumns: Boolean = false
+ var filterOutNullPartitionValues: Boolean = false
+ var tableFilter: String = ""
+ var numPartitions: Int = 100
+
+ parseArgs(args.toList)
+
+ private def parseArgs(inputArgs: List[String]): Unit = {
+ var args = inputArgs
+
+ while (args.nonEmpty) {
+ args match {
+ case "--master" :: value :: tail =>
+ master = value
+ args = tail
+
+ case "--dsdgenDir" :: value :: tail =>
+ dsdgenDir = value
+ args = tail
+
+ case "--location" :: value :: tail =>
+ location = value
+ args = tail
+
+ case "--scaleFactor" :: value :: tail =>
+ scaleFactor = toPositiveIntValue("Scale factor", value)
+ args = tail
+
+ case "--format" :: value :: tail =>
+ format = value
+ args = tail
+
+ case "--overwrite" :: tail =>
+ overwrite = true
+ args = tail
+
+ case "--partitionTables" :: tail =>
+ partitionTables = true
+ args = tail
+
+ case "--clusterByPartitionColumns" :: tail =>
+ clusterByPartitionColumns = true
+ args = tail
+
+ case "--filterOutNullPartitionValues" :: tail =>
+ filterOutNullPartitionValues = true
+ args = tail
+
+ case "--tableFilter" :: value :: tail =>
+ tableFilter = value
+ args = tail
+
+ case "--numPartitions" :: value :: tail =>
+ numPartitions = toPositiveIntValue("Number of partitions", value)
+ args = tail
+
+ case "--help" :: tail =>
+ printUsageAndExit(0)
+
+ case _ =>
+ // scalastyle:off println
+ System.err.println("Unknown/unsupported param " + args)
+ // scalastyle:on println
+ printUsageAndExit(1)
+ }
+ }
+
+ checkRequiredArguments()
+ }
+
+ private def printUsageAndExit(exitCode: Int): Unit = {
+ // scalastyle:off
+ System.err.println("""
+ |build/sbt "test:runMain <this class> [Options]"
+ |Options:
+ | --master the Spark master to use, default to local[*]
+ | --dsdgenDir location of dsdgen
+ | --location root directory of location to generate data in
+ | --scaleFactor size of the dataset to generate (in GB)
+ | --format generated data format, Parquet, ORC ...
+ | --overwrite whether to overwrite the data that is already there
+ | --partitionTables whether to create the partitioned fact tables
+ | --clusterByPartitionColumns whether to shuffle to get partitions coalesced into single files
+ | --filterOutNullPartitionValues whether to filter out the partition with NULL key value
+ | --tableFilter comma-separated list of table names to generate (e.g., store_sales,store_returns),
+ | all the tables are generated by default
+ | --numPartitions how many dsdgen partitions to run - number of input tasks
+ """.stripMargin)
+ // scalastyle:on
+ System.exit(exitCode)
+ }
+
+ private def toPositiveIntValue(name: String, v: String): Int = {
+ if (Try(v.toInt).getOrElse(-1) <= 0) {
+ // scalastyle:off println
+ System.err.println(s"$name must be a positive number")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ v.toInt
+ }
+
+ private def checkRequiredArguments(): Unit = {
+ if (dsdgenDir == null) {
+ // scalastyle:off println
+ System.err.println("Must specify a dsdgen path")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ if (location == null) {
+ // scalastyle:off println
+ System.err.println("Must specify an output location")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ }
+}
+
+/**
+ * This class generates TPCDS table data by using tpcds-kit:
+ * - https://github.com/databricks/tpcds-kit
+ *
+ * To run this:
+ * {{{
+ * build/sbt "sql/test:runMain <this class> --dsdgenDir <path> --location <path> --scaleFactor 1"
+ * }}}
+ */
+object GenTPCDSData {
+
+ def main(args: Array[String]): Unit = {
+ val config = new GenTPCDSDataConfig(args)
Review comment:
I rewrote the logic to parse options because Spark does not have the dependency of `scopt`. cc: @wangyum
https://github.com/databricks/spark-sql-perf/blob/ca4ccea3dd824597601e001d3fefe205e2bf50b0/src/main/scala/com/databricks/spark/sql/perf/tpcds/GenTPCDSData.scala#L45
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830104699
**[Test build #138110 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138110/testReport)** for PR 32243 at commit [`5ee9c1a`](https://github.com/apache/spark/commit/5ee9c1a0f87233ed1b188dddcead9199c466c5c5).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-829780747
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822973963
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137658/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830458545
**[Test build #138117 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138117/testReport)** for PR 32243 at commit [`a88fe32`](https://github.com/apache/spark/commit/a88fe32b241ec1f1a2674b69944640e1445d71cf).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-829756672
**[Test build #138095 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138095/testReport)** for PR 32243 at commit [`ef68b61`](https://github.com/apache/spark/commit/ef68b61a8d5cad890a010ec3b02459dd396c6000).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32243: [WIP][SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-829782931
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42615/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r624011869
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/tpcdsDagagen.scala
##########
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.immutable.Stream
+import scala.sys.process._
+import scala.util.Try
+
+import org.apache.spark.SparkContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.functions.{col, rpad}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
+
+// The classes in this file are basically moved from https://github.com/databricks/spark-sql-perf
+
+/**
+ * Using ProcessBuilder.lineStream produces a stream, that uses
+ * a LinkedBlockingQueue with a default capacity of Integer.MAX_VALUE.
+ *
+ * This causes OOM if the consumer cannot keep up with the producer.
+ *
+ * See scala.sys.process.ProcessBuilderImpl.lineStream
+ */
+object BlockingLineStream {
+
+ // See scala.sys.process.Streamed
+ private final class BlockingStreamed[T](
+ val process: T => Unit,
+ val done: Int => Unit,
+ val stream: () => Stream[T])
+
+ // See scala.sys.process.Streamed
+ private object BlockingStreamed {
+ // scala.process.sys.Streamed uses default of Integer.MAX_VALUE,
+ // which causes OOMs if the consumer cannot keep up with producer.
+ val maxQueueSize = 65536
+
+ def apply[T](nonzeroException: Boolean): BlockingStreamed[T] = {
+ val q = new LinkedBlockingQueue[Either[Int, T]](maxQueueSize)
+
+ def next(): Stream[T] = q.take match {
+ case Left(0) => Stream.empty
+ case Left(code) =>
+ if (nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
+ case Right(s) => Stream.cons(s, next())
+ }
+
+ new BlockingStreamed((s: T) => q put Right(s), code => q put Left(code), () => next())
+ }
+ }
+
+ // See scala.sys.process.ProcessImpl.Spawn
+ private object Spawn {
+ def apply(f: => Unit): Thread = apply(f, daemon = false)
+ def apply(f: => Unit, daemon: Boolean): Thread = {
+ val thread = new Thread() { override def run() = { f } }
+ thread.setDaemon(daemon)
+ thread.start()
+ thread
+ }
+ }
+
+ def apply(command: Seq[String]): Stream[String] = {
+ val streamed = BlockingStreamed[String](true)
+ val process = command.run(BasicIO(false, streamed.process, None))
+ Spawn(streamed.done(process.exitValue()))
+ streamed.stream()
+ }
+}
+
+class Dsdgen(dsdgenDir: String) extends Serializable {
+ private val dsdgen = s"$dsdgenDir/dsdgen"
+
+ def generate(
+ sparkContext: SparkContext,
+ tableName: String,
+ partitions: Int,
+ scaleFactor: Int): RDD[String] = {
+ val generatedData = {
+ sparkContext.parallelize(1 to partitions, partitions).flatMap { i =>
+ val localToolsDir = if (new java.io.File(dsdgen).exists) {
+ dsdgenDir
+ } else if (new java.io.File(s"/$dsdgen").exists) {
+ s"/$dsdgenDir"
+ } else {
+ throw new IllegalStateException(
+ s"Could not find dsdgen at $dsdgen or /$dsdgen. Run install")
+ }
+
+ // NOTE: RNGSEED is the RNG seed used by the data generator. Right now, it is fixed to
+ // 19620718 that is used to generate `https://github.com/maropu/spark-tpcds-sf-1`.
+ val parallel = if (partitions > 1) s"-parallel $partitions -child $i" else ""
+ val commands = Seq(
+ "bash", "-c",
+ s"cd $localToolsDir && ./dsdgen -table $tableName -filter Y -scale $scaleFactor " +
+ s"-RNGSEED 19620718 $parallel")
+ BlockingLineStream(commands)
+ }
+ }
+
+ generatedData.setName(s"$tableName, sf=$scaleFactor, strings")
+ generatedData
+ }
+}
+
+class TPCDSTables(sqlContext: SQLContext, dsdgenDir: String, scaleFactor: Int)
+ extends TPCDSSchema with Logging with Serializable {
+
+ private val dataGenerator = new Dsdgen(dsdgenDir)
+
+ private def tables: Seq[Table] = tableColumns.map { case (tableName, schemaString) =>
+ val partitionColumns = tablePartitionColumns.getOrElse(tableName, Nil)
+ .map(_.stripPrefix("`").stripSuffix("`"))
+ Table(tableName, partitionColumns, StructType.fromDDL(schemaString))
+ }.toSeq
+
+ private case class Table(name: String, partitionColumns: Seq[String], schema: StructType) {
+ def nonPartitioned: Table = {
+ Table(name, Nil, schema)
+ }
+
+ private def df(numPartition: Int) = {
+ val generatedData = dataGenerator.generate(
+ sqlContext.sparkContext, name, numPartition, scaleFactor)
+ val rows = generatedData.mapPartitions { iter =>
+ iter.map { l =>
+ val values = l.split("\\|", -1).dropRight(1).map { v =>
+ if (v.equals("")) {
+ // If the string value is an empty string, we turn it to a null
+ null
+ } else {
+ v
+ }
+ }
+ Row.fromSeq(values)
+ }
+ }
+
+ val stringData =
+ sqlContext.createDataFrame(
+ rows,
+ StructType(schema.fields.map(f => StructField(f.name, StringType))))
+
+ val convertedData = {
+ val columns = schema.fields.map { f =>
+ val c = f.dataType match {
+ // Needs right-padding for char types
+ case CharType(n) => rpad(Column(f.name), n, " ")
+ // Don't need a cast for varchar types
+ case _: VarcharType => col(f.name)
+ case _ => col(f.name).cast(f.dataType)
+ }
+ c.as(f.name)
+ }
+ stringData.select(columns: _*)
+ }
+
+ convertedData
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ numPartitions: Int): Unit = {
+ val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Ignore
+
+ val data = df(numPartitions)
+ val tempTableName = s"${name}_text"
+ data.createOrReplaceTempView(tempTableName)
+
+ val writer = if (partitionColumns.nonEmpty) {
+ if (clusterByPartitionColumns) {
+ val columnString = data.schema.fields.map { field =>
+ field.name
+ }.mkString(",")
+ val partitionColumnString = partitionColumns.mkString(",")
+ val predicates = if (filterOutNullPartitionValues) {
+ partitionColumns.map(col => s"$col IS NOT NULL").mkString("WHERE ", " AND ", "")
+ } else {
+ ""
+ }
+
+ val query =
+ s"""
+ |SELECT
+ | $columnString
+ |FROM
+ | $tempTableName
+ |$predicates
+ |DISTRIBUTE BY
+ | $partitionColumnString
+ """.stripMargin
+ val grouped = sqlContext.sql(query)
+ logInfo(s"Pre-clustering with partitioning columns with query $query.")
+ grouped.write
+ } else {
+ data.write
+ }
+ } else {
+ // treat non-partitioned tables as "one partition" that we want to coalesce
+ if (clusterByPartitionColumns) {
+ // in case data has more than maxRecordsPerFile, split into multiple writers to improve
+ // datagen speed files will be truncated to maxRecordsPerFile value, so the final
+ // result will be the same.
+ val numRows = data.count
+ val maxRecordPerFile = Try {
+ sqlContext.getConf("spark.sql.files.maxRecordsPerFile").toInt
+ }.getOrElse(0)
+
+ if (maxRecordPerFile > 0 && numRows > maxRecordPerFile) {
+ val numFiles = (numRows.toDouble/maxRecordPerFile).ceil.toInt
+ logInfo(s"Coalescing into $numFiles files")
+ data.coalesce(numFiles).write
+ } else {
+ data.coalesce(1).write
+ }
+ } else {
+ data.write
+ }
+ }
+ writer.format(format).mode(mode)
+ if (partitionColumns.nonEmpty) {
+ writer.partitionBy(partitionColumns: _*)
+ }
+ logInfo(s"Generating table $name in database to $location with save mode $mode.")
+ writer.save(location)
+ sqlContext.dropTempTable(tempTableName)
+ }
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ partitionTables: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ tableFilter: String = "",
+ numPartitions: Int = 100): Unit = {
+ var tablesToBeGenerated = if (partitionTables) {
+ tables
+ } else {
+ tables.map(_.nonPartitioned)
+ }
+
+ if (!tableFilter.isEmpty) {
+ tablesToBeGenerated = tablesToBeGenerated.filter(_.name == tableFilter)
+ if (tablesToBeGenerated.isEmpty) {
+ throw new RuntimeException("Bad table name filter: " + tableFilter)
+ }
+ }
+
+ tablesToBeGenerated.foreach { table =>
+ val tableLocation = s"$location/${table.name}"
+ table.genData(tableLocation, format, overwrite, clusterByPartitionColumns,
+ filterOutNullPartitionValues, numPartitions)
+ }
+ }
+}
+
+class GenTPCDSDataConfig(args: Array[String]) {
+ var master: String = "local[*]"
+ var dsdgenDir: String = null
+ var location: String = null
+ var scaleFactor: Int = 1
+ var format: String = "parquet"
+ var overwrite: Boolean = false
+ var partitionTables: Boolean = false
+ var clusterByPartitionColumns: Boolean = false
+ var filterOutNullPartitionValues: Boolean = false
+ var tableFilter: String = ""
+ var numPartitions: Int = 100
+
+ parseArgs(args.toList)
+
+ private def parseArgs(inputArgs: List[String]): Unit = {
+ var args = inputArgs
+
+ while (args.nonEmpty) {
+ args match {
+ case "--master" :: value :: tail =>
+ master = value
+ args = tail
+
+ case "--dsdgenDir" :: value :: tail =>
+ dsdgenDir = value
+ args = tail
+
+ case "--location" :: value :: tail =>
+ location = value
+ args = tail
+
+ case "--scaleFactor" :: value :: tail =>
+ scaleFactor = toPositiveIntValue("Scale factor", value)
+ args = tail
+
+ case "--format" :: value :: tail =>
+ format = value
+ args = tail
+
+ case "--overwrite" :: tail =>
+ overwrite = true
+ args = tail
+
+ case "--partitionTables" :: tail =>
+ partitionTables = true
+ args = tail
+
+ case "--clusterByPartitionColumns" :: tail =>
+ clusterByPartitionColumns = true
+ args = tail
+
+ case "--filterOutNullPartitionValues" :: tail =>
+ filterOutNullPartitionValues = true
+ args = tail
+
+ case "--tableFilter" :: value :: tail =>
+ tableFilter = value
+ args = tail
+
+ case "--numPartitions" :: value :: tail =>
+ numPartitions = toPositiveIntValue("Number of partitions", value)
+ args = tail
+
+ case "--help" :: tail =>
+ printUsageAndExit(0)
+
+ case _ =>
+ // scalastyle:off println
+ System.err.println("Unknown/unsupported param " + args)
+ // scalastyle:on println
+ printUsageAndExit(1)
+ }
+ }
+
+ checkRequiredArguments()
+ }
+
+ private def printUsageAndExit(exitCode: Int): Unit = {
+ // scalastyle:off
+ System.err.println("""
+ |build/sbt "test:runMain <this class> [Options]"
+ |Options:
+ | --master the Spark master to use, default to local[*]
+ | --dsdgenDir location of dsdgen
Review comment:
Is this the only difference from your repo?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yaooqinn commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616288139
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q23b.sql.out
##########
@@ -3,7 +3,4 @@
-- !query schema
struct<c_last_name:string,c_first_name:string,sales:decimal(28,2)>
-- !query output
-NULL Robert 598.86
-Brown Monika 6031.52
-Collins Gordon 727.57
-Green Jesse 9672.96
+
Review comment:
Does the empty result set look right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] npoggi commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
npoggi commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-823864629
Glad to see this update happening. Ideally, if we can store the [compressed] sf1 scale factor in a public server, there shouldn't be much of a need for compiling `dsdgen` and the generation. I see in the PR that there is some caching already.
With @HyukjinKwon we need to update `spark-sql-perf` to the latest TPC-DS spec and query templates. @gatorsmile let's coordinate that effort.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r624230505
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/tpcdsDagagen.scala
##########
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.immutable.Stream
+import scala.sys.process._
+import scala.util.Try
+
+import org.apache.spark.SparkContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.functions.{col, rpad}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
+
+// The classes in this file are basically moved from https://github.com/databricks/spark-sql-perf
+
+/**
+ * Using ProcessBuilder.lineStream produces a stream, that uses
+ * a LinkedBlockingQueue with a default capacity of Integer.MAX_VALUE.
+ *
+ * This causes OOM if the consumer cannot keep up with the producer.
+ *
+ * See scala.sys.process.ProcessBuilderImpl.lineStream
+ */
+object BlockingLineStream {
+
+ // See scala.sys.process.Streamed
+ private final class BlockingStreamed[T](
+ val process: T => Unit,
+ val done: Int => Unit,
+ val stream: () => Stream[T])
+
+ // See scala.sys.process.Streamed
+ private object BlockingStreamed {
+ // scala.process.sys.Streamed uses default of Integer.MAX_VALUE,
+ // which causes OOMs if the consumer cannot keep up with producer.
+ val maxQueueSize = 65536
+
+ def apply[T](nonzeroException: Boolean): BlockingStreamed[T] = {
+ val q = new LinkedBlockingQueue[Either[Int, T]](maxQueueSize)
+
+ def next(): Stream[T] = q.take match {
+ case Left(0) => Stream.empty
+ case Left(code) =>
+ if (nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
+ case Right(s) => Stream.cons(s, next())
+ }
+
+ new BlockingStreamed((s: T) => q put Right(s), code => q put Left(code), () => next())
+ }
+ }
+
+ // See scala.sys.process.ProcessImpl.Spawn
+ private object Spawn {
+ def apply(f: => Unit): Thread = apply(f, daemon = false)
+ def apply(f: => Unit, daemon: Boolean): Thread = {
+ val thread = new Thread() { override def run() = { f } }
+ thread.setDaemon(daemon)
+ thread.start()
+ thread
+ }
+ }
+
+ def apply(command: Seq[String]): Stream[String] = {
+ val streamed = BlockingStreamed[String](true)
+ val process = command.run(BasicIO(false, streamed.process, None))
+ Spawn(streamed.done(process.exitValue()))
+ streamed.stream()
+ }
+}
+
+class Dsdgen(dsdgenDir: String) extends Serializable {
+ private val dsdgen = s"$dsdgenDir/dsdgen"
+
+ def generate(
+ sparkContext: SparkContext,
+ tableName: String,
+ partitions: Int,
+ scaleFactor: Int): RDD[String] = {
+ val generatedData = {
+ sparkContext.parallelize(1 to partitions, partitions).flatMap { i =>
+ val localToolsDir = if (new java.io.File(dsdgen).exists) {
+ dsdgenDir
+ } else if (new java.io.File(s"/$dsdgen").exists) {
+ s"/$dsdgenDir"
+ } else {
+ throw new IllegalStateException(
+ s"Could not find dsdgen at $dsdgen or /$dsdgen. Run install")
+ }
+
+ // NOTE: RNGSEED is the RNG seed used by the data generator. Right now, it is fixed to
+ // 19620718 that is used to generate `https://github.com/maropu/spark-tpcds-sf-1`.
+ val parallel = if (partitions > 1) s"-parallel $partitions -child $i" else ""
+ val commands = Seq(
+ "bash", "-c",
+ s"cd $localToolsDir && ./dsdgen -table $tableName -filter Y -scale $scaleFactor " +
+ s"-RNGSEED 19620718 $parallel")
+ BlockingLineStream(commands)
+ }
+ }
+
+ generatedData.setName(s"$tableName, sf=$scaleFactor, strings")
+ generatedData
+ }
+}
+
+class TPCDSTables(sqlContext: SQLContext, dsdgenDir: String, scaleFactor: Int)
+ extends TPCDSSchema with Logging with Serializable {
+
+ private val dataGenerator = new Dsdgen(dsdgenDir)
+
+ private def tables: Seq[Table] = tableColumns.map { case (tableName, schemaString) =>
+ val partitionColumns = tablePartitionColumns.getOrElse(tableName, Nil)
+ .map(_.stripPrefix("`").stripSuffix("`"))
+ Table(tableName, partitionColumns, StructType.fromDDL(schemaString))
+ }.toSeq
+
+ private case class Table(name: String, partitionColumns: Seq[String], schema: StructType) {
+ def nonPartitioned: Table = {
+ Table(name, Nil, schema)
+ }
+
+ private def df(numPartition: Int) = {
+ val generatedData = dataGenerator.generate(
+ sqlContext.sparkContext, name, numPartition, scaleFactor)
+ val rows = generatedData.mapPartitions { iter =>
+ iter.map { l =>
+ val values = l.split("\\|", -1).dropRight(1).map { v =>
+ if (v.equals("")) {
+ // If the string value is an empty string, we turn it to a null
+ null
+ } else {
+ v
+ }
+ }
+ Row.fromSeq(values)
+ }
+ }
+
+ val stringData =
+ sqlContext.createDataFrame(
+ rows,
+ StructType(schema.fields.map(f => StructField(f.name, StringType))))
+
+ val convertedData = {
+ val columns = schema.fields.map { f =>
+ val c = f.dataType match {
+ // Needs right-padding for char types
+ case CharType(n) => rpad(Column(f.name), n, " ")
+ // Don't need a cast for varchar types
+ case _: VarcharType => col(f.name)
+ case _ => col(f.name).cast(f.dataType)
+ }
+ c.as(f.name)
+ }
+ stringData.select(columns: _*)
+ }
+
+ convertedData
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ numPartitions: Int): Unit = {
+ val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Ignore
+
+ val data = df(numPartitions)
+ val tempTableName = s"${name}_text"
+ data.createOrReplaceTempView(tempTableName)
+
+ val writer = if (partitionColumns.nonEmpty) {
+ if (clusterByPartitionColumns) {
+ val columnString = data.schema.fields.map { field =>
+ field.name
+ }.mkString(",")
+ val partitionColumnString = partitionColumns.mkString(",")
+ val predicates = if (filterOutNullPartitionValues) {
+ partitionColumns.map(col => s"$col IS NOT NULL").mkString("WHERE ", " AND ", "")
+ } else {
+ ""
+ }
+
+ val query =
+ s"""
+ |SELECT
+ | $columnString
+ |FROM
+ | $tempTableName
+ |$predicates
+ |DISTRIBUTE BY
+ | $partitionColumnString
+ """.stripMargin
+ val grouped = sqlContext.sql(query)
+ logInfo(s"Pre-clustering with partitioning columns with query $query.")
+ grouped.write
+ } else {
+ data.write
+ }
+ } else {
+ // treat non-partitioned tables as "one partition" that we want to coalesce
+ if (clusterByPartitionColumns) {
+ // in case data has more than maxRecordsPerFile, split into multiple writers to improve
+ // datagen speed files will be truncated to maxRecordsPerFile value, so the final
+ // result will be the same.
+ val numRows = data.count
+ val maxRecordPerFile = Try {
+ sqlContext.getConf("spark.sql.files.maxRecordsPerFile").toInt
+ }.getOrElse(0)
+
+ if (maxRecordPerFile > 0 && numRows > maxRecordPerFile) {
+ val numFiles = (numRows.toDouble/maxRecordPerFile).ceil.toInt
+ logInfo(s"Coalescing into $numFiles files")
+ data.coalesce(numFiles).write
+ } else {
+ data.coalesce(1).write
+ }
+ } else {
+ data.write
+ }
+ }
+ writer.format(format).mode(mode)
+ if (partitionColumns.nonEmpty) {
+ writer.partitionBy(partitionColumns: _*)
+ }
+ logInfo(s"Generating table $name in database to $location with save mode $mode.")
+ writer.save(location)
+ sqlContext.dropTempTable(tempTableName)
+ }
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ partitionTables: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ tableFilter: String = "",
+ numPartitions: Int = 100): Unit = {
+ var tablesToBeGenerated = if (partitionTables) {
+ tables
+ } else {
+ tables.map(_.nonPartitioned)
+ }
+
+ if (!tableFilter.isEmpty) {
+ tablesToBeGenerated = tablesToBeGenerated.filter(_.name == tableFilter)
+ if (tablesToBeGenerated.isEmpty) {
+ throw new RuntimeException("Bad table name filter: " + tableFilter)
+ }
+ }
+
+ tablesToBeGenerated.foreach { table =>
+ val tableLocation = s"$location/${table.name}"
+ table.genData(tableLocation, format, overwrite, clusterByPartitionColumns,
+ filterOutNullPartitionValues, numPartitions)
+ }
+ }
+}
+
+class GenTPCDSDataConfig(args: Array[String]) {
+ var master: String = "local[*]"
+ var dsdgenDir: String = null
+ var location: String = null
+ var scaleFactor: Int = 1
+ var format: String = "parquet"
+ var overwrite: Boolean = false
+ var partitionTables: Boolean = false
+ var clusterByPartitionColumns: Boolean = false
+ var filterOutNullPartitionValues: Boolean = false
+ var tableFilter: String = ""
+ var numPartitions: Int = 100
+
+ parseArgs(args.toList)
+
+ private def parseArgs(inputArgs: List[String]): Unit = {
+ var args = inputArgs
+
+ while (args.nonEmpty) {
+ args match {
+ case "--master" :: value :: tail =>
+ master = value
+ args = tail
+
+ case "--dsdgenDir" :: value :: tail =>
+ dsdgenDir = value
+ args = tail
+
+ case "--location" :: value :: tail =>
+ location = value
+ args = tail
+
+ case "--scaleFactor" :: value :: tail =>
+ scaleFactor = toPositiveIntValue("Scale factor", value)
+ args = tail
+
+ case "--format" :: value :: tail =>
+ format = value
+ args = tail
+
+ case "--overwrite" :: tail =>
+ overwrite = true
+ args = tail
+
+ case "--partitionTables" :: tail =>
+ partitionTables = true
+ args = tail
+
+ case "--clusterByPartitionColumns" :: tail =>
+ clusterByPartitionColumns = true
+ args = tail
+
+ case "--filterOutNullPartitionValues" :: tail =>
+ filterOutNullPartitionValues = true
+ args = tail
+
+ case "--tableFilter" :: value :: tail =>
+ tableFilter = value
+ args = tail
+
+ case "--numPartitions" :: value :: tail =>
+ numPartitions = toPositiveIntValue("Number of partitions", value)
+ args = tail
+
+ case "--help" :: tail =>
+ printUsageAndExit(0)
+
+ case _ =>
+ // scalastyle:off println
+ System.err.println("Unknown/unsupported param " + args)
+ // scalastyle:on println
+ printUsageAndExit(1)
+ }
+ }
+
+ checkRequiredArguments()
+ }
+
+ private def printUsageAndExit(exitCode: Int): Unit = {
+ // scalastyle:off
+ System.err.println("""
+ |build/sbt "test:runMain <this class> [Options]"
+ |Options:
+ | --master the Spark master to use, default to local[*]
+ | --dsdgenDir location of dsdgen
+ | --location root directory of location to generate data in
+ | --scaleFactor size of the dataset to generate (in GB)
+ | --format generated data format, Parquet, ORC ...
+ | --overwrite whether to overwrite the data that is already there
+ | --partitionTables whether to create the partitioned fact tables
+ | --clusterByPartitionColumns whether to shuffle to get partitions coalesced into single files
+ | --filterOutNullPartitionValues whether to filter out the partition with NULL key value
+ | --tableFilter comma-separated list of table names to generate (e.g., store_sales,store_returns),
+ | all the tables are generated by default
+ | --numPartitions how many dsdgen partitions to run - number of input tasks
+ """.stripMargin)
+ // scalastyle:on
+ System.exit(exitCode)
+ }
+
+ private def toPositiveIntValue(name: String, v: String): Int = {
+ if (Try(v.toInt).getOrElse(-1) <= 0) {
+ // scalastyle:off println
+ System.err.println(s"$name must be a positive number")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ v.toInt
+ }
+
+ private def checkRequiredArguments(): Unit = {
+ if (dsdgenDir == null) {
+ // scalastyle:off println
+ System.err.println("Must specify a dsdgen path")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ if (location == null) {
+ // scalastyle:off println
+ System.err.println("Must specify an output location")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ }
+}
+
+/**
+ * This class generates TPCDS table data by using tpcds-kit:
+ * - https://github.com/databricks/tpcds-kit
+ *
+ * To run this:
+ * {{{
+ * build/sbt "sql/test:runMain <this class> --dsdgenDir <path> --location <path> --scaleFactor 1"
+ * }}}
+ */
+object GenTPCDSData {
+
+ def main(args: Array[String]): Unit = {
+ val config = new GenTPCDSDataConfig(args)
+
+ val spark = SparkSession
+ .builder()
+ .appName(getClass.getName)
+ .master(config.master)
+ .getOrCreate()
+
+ val tables = new TPCDSTables(
+ spark.sqlContext,
+ dsdgenDir = config.dsdgenDir,
+ scaleFactor = config.scaleFactor)
+
+ tables.genData(
+ location = config.location,
+ format = config.format,
+ overwrite = config.overwrite,
+ partitionTables = config.partitionTables,
+ clusterByPartitionColumns = config.clusterByPartitionColumns,
+ filterOutNullPartitionValues = config.filterOutNullPartitionValues,
+ tableFilter = config.tableFilter,
+ numPartitions = config.numPartitions)
Review comment:
Oh, I forgot it. Thank you.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-822905978
WDYT? If no one disagrees with this, I'll file jira for it. @juliuszsompolski @npoggi @wangyum @HyukjinKwon @yaooqinn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r624011221
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/tpcdsDagagen.scala
##########
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.immutable.Stream
+import scala.sys.process._
+import scala.util.Try
+
+import org.apache.spark.SparkContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.functions.{col, rpad}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
+
+// The classes in this file are basically moved from https://github.com/databricks/spark-sql-perf
+
+/**
+ * Using ProcessBuilder.lineStream produces a stream, that uses
+ * a LinkedBlockingQueue with a default capacity of Integer.MAX_VALUE.
+ *
+ * This causes OOM if the consumer cannot keep up with the producer.
+ *
+ * See scala.sys.process.ProcessBuilderImpl.lineStream
+ */
+object BlockingLineStream {
+
+ // See scala.sys.process.Streamed
+ private final class BlockingStreamed[T](
+ val process: T => Unit,
+ val done: Int => Unit,
+ val stream: () => Stream[T])
+
+ // See scala.sys.process.Streamed
+ private object BlockingStreamed {
+ // scala.process.sys.Streamed uses default of Integer.MAX_VALUE,
+ // which causes OOMs if the consumer cannot keep up with producer.
+ val maxQueueSize = 65536
+
+ def apply[T](nonzeroException: Boolean): BlockingStreamed[T] = {
+ val q = new LinkedBlockingQueue[Either[Int, T]](maxQueueSize)
+
+ def next(): Stream[T] = q.take match {
+ case Left(0) => Stream.empty
+ case Left(code) =>
+ if (nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
+ case Right(s) => Stream.cons(s, next())
+ }
+
+ new BlockingStreamed((s: T) => q put Right(s), code => q put Left(code), () => next())
+ }
+ }
+
+ // See scala.sys.process.ProcessImpl.Spawn
+ private object Spawn {
+ def apply(f: => Unit): Thread = apply(f, daemon = false)
+ def apply(f: => Unit, daemon: Boolean): Thread = {
+ val thread = new Thread() { override def run() = { f } }
+ thread.setDaemon(daemon)
+ thread.start()
+ thread
+ }
+ }
+
+ def apply(command: Seq[String]): Stream[String] = {
+ val streamed = BlockingStreamed[String](true)
+ val process = command.run(BasicIO(false, streamed.process, None))
+ Spawn(streamed.done(process.exitValue()))
+ streamed.stream()
+ }
+}
+
+class Dsdgen(dsdgenDir: String) extends Serializable {
+ private val dsdgen = s"$dsdgenDir/dsdgen"
+
+ def generate(
+ sparkContext: SparkContext,
+ tableName: String,
+ partitions: Int,
+ scaleFactor: Int): RDD[String] = {
+ val generatedData = {
+ sparkContext.parallelize(1 to partitions, partitions).flatMap { i =>
+ val localToolsDir = if (new java.io.File(dsdgen).exists) {
+ dsdgenDir
+ } else if (new java.io.File(s"/$dsdgen").exists) {
+ s"/$dsdgenDir"
+ } else {
+ throw new IllegalStateException(
+ s"Could not find dsdgen at $dsdgen or /$dsdgen. Run install")
+ }
+
+ // NOTE: RNGSEED is the RNG seed used by the data generator. Right now, it is fixed to
+ // 19620718 that is used to generate `https://github.com/maropu/spark-tpcds-sf-1`.
+ val parallel = if (partitions > 1) s"-parallel $partitions -child $i" else ""
+ val commands = Seq(
+ "bash", "-c",
+ s"cd $localToolsDir && ./dsdgen -table $tableName -filter Y -scale $scaleFactor " +
+ s"-RNGSEED 19620718 $parallel")
+ BlockingLineStream(commands)
+ }
+ }
+
+ generatedData.setName(s"$tableName, sf=$scaleFactor, strings")
+ generatedData
+ }
+}
+
+class TPCDSTables(sqlContext: SQLContext, dsdgenDir: String, scaleFactor: Int)
+ extends TPCDSSchema with Logging with Serializable {
+
+ private val dataGenerator = new Dsdgen(dsdgenDir)
+
+ private def tables: Seq[Table] = tableColumns.map { case (tableName, schemaString) =>
+ val partitionColumns = tablePartitionColumns.getOrElse(tableName, Nil)
+ .map(_.stripPrefix("`").stripSuffix("`"))
+ Table(tableName, partitionColumns, StructType.fromDDL(schemaString))
+ }.toSeq
+
+ private case class Table(name: String, partitionColumns: Seq[String], schema: StructType) {
+ def nonPartitioned: Table = {
+ Table(name, Nil, schema)
+ }
+
+ private def df(numPartition: Int) = {
+ val generatedData = dataGenerator.generate(
+ sqlContext.sparkContext, name, numPartition, scaleFactor)
+ val rows = generatedData.mapPartitions { iter =>
+ iter.map { l =>
+ val values = l.split("\\|", -1).dropRight(1).map { v =>
+ if (v.equals("")) {
+ // If the string value is an empty string, we turn it to a null
+ null
+ } else {
+ v
+ }
+ }
+ Row.fromSeq(values)
+ }
+ }
+
+ val stringData =
+ sqlContext.createDataFrame(
+ rows,
+ StructType(schema.fields.map(f => StructField(f.name, StringType))))
+
+ val convertedData = {
+ val columns = schema.fields.map { f =>
+ val c = f.dataType match {
+ // Needs right-padding for char types
+ case CharType(n) => rpad(Column(f.name), n, " ")
+ // Don't need a cast for varchar types
+ case _: VarcharType => col(f.name)
+ case _ => col(f.name).cast(f.dataType)
+ }
+ c.as(f.name)
+ }
+ stringData.select(columns: _*)
+ }
+
+ convertedData
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ numPartitions: Int): Unit = {
+ val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Ignore
+
+ val data = df(numPartitions)
+ val tempTableName = s"${name}_text"
+ data.createOrReplaceTempView(tempTableName)
+
+ val writer = if (partitionColumns.nonEmpty) {
+ if (clusterByPartitionColumns) {
+ val columnString = data.schema.fields.map { field =>
+ field.name
+ }.mkString(",")
+ val partitionColumnString = partitionColumns.mkString(",")
+ val predicates = if (filterOutNullPartitionValues) {
+ partitionColumns.map(col => s"$col IS NOT NULL").mkString("WHERE ", " AND ", "")
+ } else {
+ ""
+ }
+
+ val query =
+ s"""
+ |SELECT
+ | $columnString
+ |FROM
+ | $tempTableName
+ |$predicates
+ |DISTRIBUTE BY
+ | $partitionColumnString
+ """.stripMargin
+ val grouped = sqlContext.sql(query)
+ logInfo(s"Pre-clustering with partitioning columns with query $query.")
+ grouped.write
+ } else {
+ data.write
+ }
+ } else {
+ // treat non-partitioned tables as "one partition" that we want to coalesce
+ if (clusterByPartitionColumns) {
+ // in case data has more than maxRecordsPerFile, split into multiple writers to improve
+ // datagen speed files will be truncated to maxRecordsPerFile value, so the final
+ // result will be the same.
+ val numRows = data.count
+ val maxRecordPerFile = Try {
+ sqlContext.getConf("spark.sql.files.maxRecordsPerFile").toInt
+ }.getOrElse(0)
+
+ if (maxRecordPerFile > 0 && numRows > maxRecordPerFile) {
+ val numFiles = (numRows.toDouble/maxRecordPerFile).ceil.toInt
+ logInfo(s"Coalescing into $numFiles files")
+ data.coalesce(numFiles).write
+ } else {
+ data.coalesce(1).write
+ }
+ } else {
+ data.write
+ }
+ }
+ writer.format(format).mode(mode)
+ if (partitionColumns.nonEmpty) {
+ writer.partitionBy(partitionColumns: _*)
+ }
+ logInfo(s"Generating table $name in database to $location with save mode $mode.")
+ writer.save(location)
+ sqlContext.dropTempTable(tempTableName)
+ }
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ partitionTables: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ tableFilter: String = "",
+ numPartitions: Int = 100): Unit = {
+ var tablesToBeGenerated = if (partitionTables) {
+ tables
+ } else {
+ tables.map(_.nonPartitioned)
+ }
+
+ if (!tableFilter.isEmpty) {
+ tablesToBeGenerated = tablesToBeGenerated.filter(_.name == tableFilter)
+ if (tablesToBeGenerated.isEmpty) {
+ throw new RuntimeException("Bad table name filter: " + tableFilter)
+ }
+ }
+
+ tablesToBeGenerated.foreach { table =>
+ val tableLocation = s"$location/${table.name}"
+ table.genData(tableLocation, format, overwrite, clusterByPartitionColumns,
+ filterOutNullPartitionValues, numPartitions)
+ }
+ }
+}
+
+class GenTPCDSDataConfig(args: Array[String]) {
+ var master: String = "local[*]"
+ var dsdgenDir: String = null
+ var location: String = null
+ var scaleFactor: Int = 1
+ var format: String = "parquet"
+ var overwrite: Boolean = false
+ var partitionTables: Boolean = false
+ var clusterByPartitionColumns: Boolean = false
+ var filterOutNullPartitionValues: Boolean = false
+ var tableFilter: String = ""
+ var numPartitions: Int = 100
+
+ parseArgs(args.toList)
+
+ private def parseArgs(inputArgs: List[String]): Unit = {
+ var args = inputArgs
+
+ while (args.nonEmpty) {
+ args match {
+ case "--master" :: value :: tail =>
+ master = value
+ args = tail
+
+ case "--dsdgenDir" :: value :: tail =>
+ dsdgenDir = value
+ args = tail
+
+ case "--location" :: value :: tail =>
+ location = value
+ args = tail
+
+ case "--scaleFactor" :: value :: tail =>
+ scaleFactor = toPositiveIntValue("Scale factor", value)
+ args = tail
+
+ case "--format" :: value :: tail =>
+ format = value
+ args = tail
+
+ case "--overwrite" :: tail =>
+ overwrite = true
+ args = tail
+
+ case "--partitionTables" :: tail =>
+ partitionTables = true
+ args = tail
+
+ case "--clusterByPartitionColumns" :: tail =>
+ clusterByPartitionColumns = true
+ args = tail
+
+ case "--filterOutNullPartitionValues" :: tail =>
+ filterOutNullPartitionValues = true
+ args = tail
+
+ case "--tableFilter" :: value :: tail =>
+ tableFilter = value
+ args = tail
+
+ case "--numPartitions" :: value :: tail =>
+ numPartitions = toPositiveIntValue("Number of partitions", value)
+ args = tail
+
+ case "--help" :: tail =>
+ printUsageAndExit(0)
+
+ case _ =>
+ // scalastyle:off println
+ System.err.println("Unknown/unsupported param " + args)
+ // scalastyle:on println
+ printUsageAndExit(1)
+ }
+ }
+
+ checkRequiredArguments()
+ }
+
+ private def printUsageAndExit(exitCode: Int): Unit = {
+ // scalastyle:off
+ System.err.println("""
+ |build/sbt "test:runMain <this class> [Options]"
+ |Options:
+ | --master the Spark master to use, default to local[*]
+ | --dsdgenDir location of dsdgen
+ | --location root directory of location to generate data in
+ | --scaleFactor size of the dataset to generate (in GB)
+ | --format generated data format, Parquet, ORC ...
+ | --overwrite whether to overwrite the data that is already there
+ | --partitionTables whether to create the partitioned fact tables
+ | --clusterByPartitionColumns whether to shuffle to get partitions coalesced into single files
+ | --filterOutNullPartitionValues whether to filter out the partition with NULL key value
+ | --tableFilter comma-separated list of table names to generate (e.g., store_sales,store_returns),
+ | all the tables are generated by default
+ | --numPartitions how many dsdgen partitions to run - number of input tasks
+ """.stripMargin)
+ // scalastyle:on
+ System.exit(exitCode)
+ }
+
+ private def toPositiveIntValue(name: String, v: String): Int = {
+ if (Try(v.toInt).getOrElse(-1) <= 0) {
+ // scalastyle:off println
+ System.err.println(s"$name must be a positive number")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ v.toInt
+ }
+
+ private def checkRequiredArguments(): Unit = {
+ if (dsdgenDir == null) {
+ // scalastyle:off println
+ System.err.println("Must specify a dsdgen path")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ if (location == null) {
+ // scalastyle:off println
+ System.err.println("Must specify an output location")
+ // scalastyle:on println
+ printUsageAndExit(-1)
+ }
+ }
+}
+
+/**
+ * This class generates TPCDS table data by using tpcds-kit:
+ * - https://github.com/databricks/tpcds-kit
+ *
+ * To run this:
+ * {{{
+ * build/sbt "sql/test:runMain <this class> --dsdgenDir <path> --location <path> --scaleFactor 1"
+ * }}}
+ */
+object GenTPCDSData {
Review comment:
Shall we rename the file `tpcdsDatagen.scala` -> `GenTPCDSData.scala`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [WIP][SPARK-XXXXX][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r616359569
##########
File path: sql/core/src/test/resources/tpcds-query-results/v1_4/q1.sql.out
##########
@@ -3,103 +3,103 @@
-- !query schema
struct<c_customer_id:string>
-- !query output
-AAAAAAAAAAABBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAADBAAA
-AAAAAAAAAAAKAAAA
-AAAAAAAAAABDAAAA
-AAAAAAAAAABHBAAA
-AAAAAAAAAABLAAAA
-AAAAAAAAAABMAAAA
-AAAAAAAAAACHAAAA
-AAAAAAAAAACMAAAA
-AAAAAAAAAADDAAAA
-AAAAAAAAAADGAAAA
-AAAAAAAAAADGBAAA
-AAAAAAAAAADGBAAA
+AAAAAAAAAAABAAAA
Review comment:
yea, sgtm. We can split it by checking out my repo instead: https://github.com/maropu/spark-tpcds-datagen/tree/master/thirdparty/tpcds-kit/tools
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32243:
URL: https://github.com/apache/spark/pull/32243#discussion_r624230272
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/tpcdsDagagen.scala
##########
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.immutable.Stream
+import scala.sys.process._
+import scala.util.Try
+
+import org.apache.spark.SparkContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.functions.{col, rpad}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
+
+// The classes in this file are basically moved from https://github.com/databricks/spark-sql-perf
+
+/**
+ * Using ProcessBuilder.lineStream produces a stream, that uses
+ * a LinkedBlockingQueue with a default capacity of Integer.MAX_VALUE.
+ *
+ * This causes OOM if the consumer cannot keep up with the producer.
+ *
+ * See scala.sys.process.ProcessBuilderImpl.lineStream
+ */
+object BlockingLineStream {
+
+ // See scala.sys.process.Streamed
+ private final class BlockingStreamed[T](
+ val process: T => Unit,
+ val done: Int => Unit,
+ val stream: () => Stream[T])
+
+ // See scala.sys.process.Streamed
+ private object BlockingStreamed {
+ // scala.process.sys.Streamed uses default of Integer.MAX_VALUE,
+ // which causes OOMs if the consumer cannot keep up with producer.
+ val maxQueueSize = 65536
+
+ def apply[T](nonzeroException: Boolean): BlockingStreamed[T] = {
+ val q = new LinkedBlockingQueue[Either[Int, T]](maxQueueSize)
+
+ def next(): Stream[T] = q.take match {
+ case Left(0) => Stream.empty
+ case Left(code) =>
+ if (nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
+ case Right(s) => Stream.cons(s, next())
+ }
+
+ new BlockingStreamed((s: T) => q put Right(s), code => q put Left(code), () => next())
+ }
+ }
+
+ // See scala.sys.process.ProcessImpl.Spawn
+ private object Spawn {
+ def apply(f: => Unit): Thread = apply(f, daemon = false)
+ def apply(f: => Unit, daemon: Boolean): Thread = {
+ val thread = new Thread() { override def run() = { f } }
+ thread.setDaemon(daemon)
+ thread.start()
+ thread
+ }
+ }
+
+ def apply(command: Seq[String]): Stream[String] = {
+ val streamed = BlockingStreamed[String](true)
+ val process = command.run(BasicIO(false, streamed.process, None))
+ Spawn(streamed.done(process.exitValue()))
+ streamed.stream()
+ }
+}
+
+class Dsdgen(dsdgenDir: String) extends Serializable {
+ private val dsdgen = s"$dsdgenDir/dsdgen"
+
+ def generate(
+ sparkContext: SparkContext,
+ tableName: String,
+ partitions: Int,
+ scaleFactor: Int): RDD[String] = {
+ val generatedData = {
+ sparkContext.parallelize(1 to partitions, partitions).flatMap { i =>
+ val localToolsDir = if (new java.io.File(dsdgen).exists) {
+ dsdgenDir
+ } else if (new java.io.File(s"/$dsdgen").exists) {
+ s"/$dsdgenDir"
+ } else {
+ throw new IllegalStateException(
+ s"Could not find dsdgen at $dsdgen or /$dsdgen. Run install")
+ }
+
+ // NOTE: RNGSEED is the RNG seed used by the data generator. Right now, it is fixed to
+ // 19620718 that is used to generate `https://github.com/maropu/spark-tpcds-sf-1`.
+ val parallel = if (partitions > 1) s"-parallel $partitions -child $i" else ""
+ val commands = Seq(
+ "bash", "-c",
+ s"cd $localToolsDir && ./dsdgen -table $tableName -filter Y -scale $scaleFactor " +
+ s"-RNGSEED 19620718 $parallel")
+ BlockingLineStream(commands)
+ }
+ }
+
+ generatedData.setName(s"$tableName, sf=$scaleFactor, strings")
+ generatedData
+ }
+}
+
+class TPCDSTables(sqlContext: SQLContext, dsdgenDir: String, scaleFactor: Int)
+ extends TPCDSSchema with Logging with Serializable {
+
+ private val dataGenerator = new Dsdgen(dsdgenDir)
+
+ private def tables: Seq[Table] = tableColumns.map { case (tableName, schemaString) =>
+ val partitionColumns = tablePartitionColumns.getOrElse(tableName, Nil)
+ .map(_.stripPrefix("`").stripSuffix("`"))
+ Table(tableName, partitionColumns, StructType.fromDDL(schemaString))
+ }.toSeq
+
+ private case class Table(name: String, partitionColumns: Seq[String], schema: StructType) {
+ def nonPartitioned: Table = {
+ Table(name, Nil, schema)
+ }
+
+ private def df(numPartition: Int) = {
+ val generatedData = dataGenerator.generate(
+ sqlContext.sparkContext, name, numPartition, scaleFactor)
+ val rows = generatedData.mapPartitions { iter =>
+ iter.map { l =>
+ val values = l.split("\\|", -1).dropRight(1).map { v =>
+ if (v.equals("")) {
+ // If the string value is an empty string, we turn it to a null
+ null
+ } else {
+ v
+ }
+ }
+ Row.fromSeq(values)
+ }
+ }
+
+ val stringData =
+ sqlContext.createDataFrame(
+ rows,
+ StructType(schema.fields.map(f => StructField(f.name, StringType))))
+
+ val convertedData = {
+ val columns = schema.fields.map { f =>
+ val c = f.dataType match {
+ // Needs right-padding for char types
+ case CharType(n) => rpad(Column(f.name), n, " ")
+ // Don't need a cast for varchar types
+ case _: VarcharType => col(f.name)
+ case _ => col(f.name).cast(f.dataType)
+ }
+ c.as(f.name)
+ }
+ stringData.select(columns: _*)
+ }
+
+ convertedData
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ numPartitions: Int): Unit = {
+ val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Ignore
+
+ val data = df(numPartitions)
+ val tempTableName = s"${name}_text"
+ data.createOrReplaceTempView(tempTableName)
+
+ val writer = if (partitionColumns.nonEmpty) {
+ if (clusterByPartitionColumns) {
+ val columnString = data.schema.fields.map { field =>
+ field.name
+ }.mkString(",")
+ val partitionColumnString = partitionColumns.mkString(",")
+ val predicates = if (filterOutNullPartitionValues) {
+ partitionColumns.map(col => s"$col IS NOT NULL").mkString("WHERE ", " AND ", "")
+ } else {
+ ""
+ }
+
+ val query =
+ s"""
+ |SELECT
+ | $columnString
+ |FROM
+ | $tempTableName
+ |$predicates
+ |DISTRIBUTE BY
+ | $partitionColumnString
+ """.stripMargin
+ val grouped = sqlContext.sql(query)
+ logInfo(s"Pre-clustering with partitioning columns with query $query.")
+ grouped.write
+ } else {
+ data.write
+ }
+ } else {
+ // treat non-partitioned tables as "one partition" that we want to coalesce
+ if (clusterByPartitionColumns) {
+ // in case data has more than maxRecordsPerFile, split into multiple writers to improve
+ // datagen speed files will be truncated to maxRecordsPerFile value, so the final
+ // result will be the same.
+ val numRows = data.count
+ val maxRecordPerFile = Try {
+ sqlContext.getConf("spark.sql.files.maxRecordsPerFile").toInt
+ }.getOrElse(0)
+
+ if (maxRecordPerFile > 0 && numRows > maxRecordPerFile) {
+ val numFiles = (numRows.toDouble/maxRecordPerFile).ceil.toInt
+ logInfo(s"Coalescing into $numFiles files")
+ data.coalesce(numFiles).write
+ } else {
+ data.coalesce(1).write
+ }
+ } else {
+ data.write
+ }
+ }
+ writer.format(format).mode(mode)
+ if (partitionColumns.nonEmpty) {
+ writer.partitionBy(partitionColumns: _*)
+ }
+ logInfo(s"Generating table $name in database to $location with save mode $mode.")
+ writer.save(location)
+ sqlContext.dropTempTable(tempTableName)
+ }
+ }
+
+ def genData(
+ location: String,
+ format: String,
+ overwrite: Boolean,
+ partitionTables: Boolean,
+ clusterByPartitionColumns: Boolean,
+ filterOutNullPartitionValues: Boolean,
+ tableFilter: String = "",
+ numPartitions: Int = 100): Unit = {
+ var tablesToBeGenerated = if (partitionTables) {
+ tables
+ } else {
+ tables.map(_.nonPartitioned)
+ }
+
+ if (!tableFilter.isEmpty) {
+ tablesToBeGenerated = tablesToBeGenerated.filter(_.name == tableFilter)
+ if (tablesToBeGenerated.isEmpty) {
+ throw new RuntimeException("Bad table name filter: " + tableFilter)
+ }
+ }
+
+ tablesToBeGenerated.foreach { table =>
+ val tableLocation = s"$location/${table.name}"
+ table.genData(tableLocation, format, overwrite, clusterByPartitionColumns,
+ filterOutNullPartitionValues, numPartitions)
+ }
+ }
+}
+
+class GenTPCDSDataConfig(args: Array[String]) {
+ var master: String = "local[*]"
+ var dsdgenDir: String = null
+ var location: String = null
+ var scaleFactor: Int = 1
+ var format: String = "parquet"
+ var overwrite: Boolean = false
+ var partitionTables: Boolean = false
+ var clusterByPartitionColumns: Boolean = false
+ var filterOutNullPartitionValues: Boolean = false
+ var tableFilter: String = ""
+ var numPartitions: Int = 100
+
+ parseArgs(args.toList)
+
+ private def parseArgs(inputArgs: List[String]): Unit = {
+ var args = inputArgs
+
+ while (args.nonEmpty) {
+ args match {
+ case "--master" :: value :: tail =>
+ master = value
+ args = tail
+
+ case "--dsdgenDir" :: value :: tail =>
+ dsdgenDir = value
+ args = tail
+
+ case "--location" :: value :: tail =>
+ location = value
+ args = tail
+
+ case "--scaleFactor" :: value :: tail =>
+ scaleFactor = toPositiveIntValue("Scale factor", value)
+ args = tail
+
+ case "--format" :: value :: tail =>
+ format = value
+ args = tail
+
+ case "--overwrite" :: tail =>
+ overwrite = true
+ args = tail
+
+ case "--partitionTables" :: tail =>
+ partitionTables = true
+ args = tail
+
+ case "--clusterByPartitionColumns" :: tail =>
+ clusterByPartitionColumns = true
+ args = tail
+
+ case "--filterOutNullPartitionValues" :: tail =>
+ filterOutNullPartitionValues = true
+ args = tail
+
+ case "--tableFilter" :: value :: tail =>
+ tableFilter = value
+ args = tail
+
+ case "--numPartitions" :: value :: tail =>
+ numPartitions = toPositiveIntValue("Number of partitions", value)
+ args = tail
+
+ case "--help" :: tail =>
+ printUsageAndExit(0)
+
+ case _ =>
+ // scalastyle:off println
+ System.err.println("Unknown/unsupported param " + args)
+ // scalastyle:on println
+ printUsageAndExit(1)
+ }
+ }
+
+ checkRequiredArguments()
+ }
+
+ private def printUsageAndExit(exitCode: Int): Unit = {
+ // scalastyle:off
+ System.err.println("""
+ |build/sbt "test:runMain <this class> [Options]"
+ |Options:
+ | --master the Spark master to use, default to local[*]
+ | --dsdgenDir location of dsdgen
Review comment:
yea, yes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830104699
**[Test build #138110 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138110/testReport)** for PR 32243 at commit [`5ee9c1a`](https://github.com/apache/spark/commit/5ee9c1a0f87233ed1b188dddcead9199c466c5c5).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830462044
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138117/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830475565
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42640/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32243: [SPARK-35192][SQL][TESTS] Port minimal TPC-DS datagen code from databricks/spark-sql-perf
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32243:
URL: https://github.com/apache/spark/pull/32243#issuecomment-830431961
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42638/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org