Spark3 tpcds setup

README.md

@@ -75,7 +75,7 @@ val databaseName = ... // name of database to create.
 val scaleFactor = ... // scaleFactor defines the size of the dataset to generate (in GB).
 val format = ... // valid spark format like parquet "parquet".
 // Run:
-val tables = new TPCDSTables(sqlContext,
+val tables = new TPCDSTables(spark, sqlContext,
     dsdgenDir = "/tmp/tpcds-kit/tools", // location of dsdgen
     scaleFactor = scaleFactor,
     useDoubleForDecimal = false, // true to replace DecimalType with DoubleType
@@ -188,4 +188,4 @@ For running parallel TPCDS streams:
 
 ### tpch_run notebook
 
-This notebook can be used to run TPCH queries.  Data needs be generated first.
+This notebook can be used to run TPCH queries.  Data needs be generated first.

src/main/notebooks/TPC-multi_datagen.scala

@@ -145,17 +145,17 @@ def getBenchmarkData(benchmark: String, scaleFactor: String) = benchmark match {
 
   case "TPCH" => (
     s"tpch_sf${scaleFactor}_${fileFormat}${dbSuffix}",
-    new TPCHTables(spark.sqlContext, dbgenDir = s"${baseDatagenFolder}/dbgen", scaleFactor = scaleFactor, useDoubleForDecimal = false, useStringForDate = false, generatorParams = Nil),
+    new TPCHTables(spark, spark.sqlContext, dbgenDir = s"${baseDatagenFolder}/dbgen", scaleFactor = scaleFactor, useDoubleForDecimal = false, useStringForDate = false, generatorParams = Nil),
     s"$baseLocation/tpch/sf${scaleFactor}_${fileFormat}")  
 
   case "TPCDS" if !TPCDSUseLegacyOptions => (
     s"tpcds_sf${scaleFactor}_${fileFormat}${dbSuffix}",
-    new TPCDSTables(spark.sqlContext, dsdgenDir = s"${baseDatagenFolder}/dsdgen", scaleFactor = scaleFactor, useDoubleForDecimal = false, useStringForDate = false),
+    new TPCDSTables(spark, spark.sqlContext, dsdgenDir = s"${baseDatagenFolder}/dsdgen", scaleFactor = scaleFactor, useDoubleForDecimal = false, useStringForDate = false),
     s"$baseLocation/tpcds-2.4/sf${scaleFactor}_${fileFormat}")
 
   case "TPCDS" if TPCDSUseLegacyOptions => (
     s"tpcds_sf${scaleFactor}_nodecimal_nodate_withnulls${dbSuffix}",
-    new TPCDSTables(spark.sqlContext, s"${baseDatagenFolder}/dsdgen", scaleFactor = scaleFactor, useDoubleForDecimal = true, useStringForDate = true),
+    new TPCDSTables(spark, spark.sqlContext, s"${baseDatagenFolder}/dsdgen", scaleFactor = scaleFactor, useDoubleForDecimal = true, useStringForDate = true),
     s"$baseLocation/tpcds/sf$scaleFactor-$fileFormat/useDecimal=false,useDate=false,filterNull=false")
 }
 

src/main/scala/com/databricks/spark/sql/perf/Tables.scala

@@ -27,7 +27,7 @@ import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
-import org.apache.spark.sql.{Row, SQLContext, SaveMode}
+import org.apache.spark.sql.{Row, SQLContext, SaveMode, SparkSession}
 
 
 /**
@@ -94,7 +94,7 @@ trait DataGenerator extends Serializable {
 }
 
 
-abstract class Tables(sqlContext: SQLContext, scaleFactor: String,
+abstract class Tables(sqlSession SparkSession, sqlContext: SQLContext, scaleFactor: String,
     useDoubleForDecimal: Boolean = false, useStringForDate: Boolean = false)
     extends Serializable {
 
@@ -254,7 +254,7 @@ abstract class Tables(sqlContext: SQLContext, scaleFactor: String,
       if (!tableExists || overwrite) {
         println(s"Creating external table $name in database $databaseName using data stored in $location.")
         log.info(s"Creating external table $name in database $databaseName using data stored in $location.")
-        sqlContext.createExternalTable(qualifiedTableName, location, format)
+        sqlSession.createTable(qualifiedTableName, location, format)
       }
       if (partitionColumns.nonEmpty && discoverPartitions) {
         println(s"Discovering partitions for table $name.")

src/main/scala/com/databricks/spark/sql/perf/tpcds/TPCDSTables.scala

@@ -22,7 +22,7 @@ import com.databricks.spark.sql.perf
 import com.databricks.spark.sql.perf.{BlockingLineStream, DataGenerator, Table, Tables}
 
 import org.apache.spark.SparkContext
-import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.{SQLContext, SparkSession}
 
 class DSDGEN(dsdgenDir: String) extends DataGenerator {
   val dsdgen = s"$dsdgenDir/dsdgen"
@@ -55,12 +55,13 @@ class DSDGEN(dsdgenDir: String) extends DataGenerator {
 
 
 class TPCDSTables(
+  sqlSession: SparkSession
   sqlContext: SQLContext,
   dsdgenDir: String,
   scaleFactor: String,
   useDoubleForDecimal: Boolean = false,
   useStringForDate: Boolean = false)
-  extends Tables(sqlContext, scaleFactor, useDoubleForDecimal, useStringForDate) {
+  extends Tables(sqlSession, sqlContext, scaleFactor, useDoubleForDecimal, useStringForDate) {
   import sqlContext.implicits._
 
   val dataGenerator = new DSDGEN(dsdgenDir)

src/main/scala/com/databricks/spark/sql/perf/tpch/TPCH.scala

@@ -22,7 +22,7 @@ import com.databricks.spark.sql.perf.ExecutionMode.CollectResults
 import org.apache.commons.io.IOUtils
 
 import org.apache.spark.SparkContext
-import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.{SQLContext, SparkSession}
 
 class DBGEN(dbgenDir: String, params: Seq[String]) extends DataGenerator {
   val dbgen = s"$dbgenDir/dbgen"
@@ -64,13 +64,14 @@ class DBGEN(dbgenDir: String, params: Seq[String]) extends DataGenerator {
 }
 
 class TPCHTables(
+    sqlSession: SparkSession
     sqlContext: SQLContext,
     dbgenDir: String,
     scaleFactor: String,
     useDoubleForDecimal: Boolean = false,
     useStringForDate: Boolean = false,
     generatorParams: Seq[String] = Nil)
-    extends Tables(sqlContext, scaleFactor, useDoubleForDecimal, useStringForDate) {
+    extends Tables(sqlSession, sqlContext, scaleFactor, useDoubleForDecimal, useStringForDate) {
   import sqlContext.implicits._
 
   val dataGenerator = new DBGEN(dbgenDir, generatorParams)