integrations/oap/dataproc/notebooks/tpcds_power_test_Dataproc.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "1b2a71ba-c8ec-4627-a732-90bc996f5ffc",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%init_spark\n",
    "launcher.jars = [\"file:///opt/benchmark-tools/spark-sql-perf/target/scala-2.12/spark-sql-perf_2.12-0.5.1-SNAPSHOT.jar\"]\n",
    "launcher.conf.set(\"spark.sql.warehouse.dir\", \"hdfs:///user/livy\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f7ccb0c8-a4b6-4cad-a09e-08ba61125c65",
   "metadata": {},
   "outputs": [],
   "source": [
    "!hadoop fs -mkdir /user/livy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "13ec09dd-fb22-4eb5-8220-1d4ca00af768",
   "metadata": {},
   "outputs": [],
   "source": [
    "val scaleFactor = \"1\"           // data scale 1GB\n",
    "val iterations = 1              // how many times to run the whole set of queries.\n",
    "val format = \"parquet\"          // support parquet or orc\n",
    "val storage = \"hdfs\"            // choose HDFS\n",
    "val bucket_name = \"/user/livy\"  // scala notebook only has the write permission of \"hdfs:///user/livy\" directory    \n",
    "val partitionTables = true      // create partition tables\n",
    "val query_filter = Seq()        // Seq() == all queries\n",
    "//val query_filter = Seq(\"q1-v2.4\", \"q2-v2.4\") // run subset of queries\n",
    "val randomizeQueries = false    // run queries in a random order. Recommended for parallel runs.\n",
    "\n",
    "// detailed results will be written as JSON to this location.\n",
    "var resultLocation = s\"${storage}://${bucket_name}/results/tpcds_${format}/${scaleFactor}/\"\n",
    "var databaseName = s\"tpcds_${format}_scale_${scaleFactor}_db\"\n",
    "val use_arrow = false            // when you want to use gazella_plugin to run TPC-DS, you need to set it true.\n",
    "\n",
    "if (use_arrow){\n",
    "    val data_path= s\"${storage}://${bucket_name}/datagen/tpcds_${format}/${scaleFactor}\"\n",
    "    resultLocation = s\"${storage}://${bucket_name}/results/tpcds_arrow/${scaleFactor}/\"\n",
    "    databaseName = s\"tpcds_arrow_scale_${scaleFactor}_db\"\n",
    "    val tables = Seq(\"call_center\", \"catalog_page\", \"catalog_returns\", \"catalog_sales\", \"customer\", \"customer_address\", \"customer_demographics\", \"date_dim\", \"household_demographics\", \"income_band\", \"inventory\", \"item\", \"promotion\", \"reason\", \"ship_mode\", \"store\", \"store_returns\", \"store_sales\", \"time_dim\", \"warehouse\", \"web_page\", \"web_returns\", \"web_sales\", \"web_site\")\n",
    "    if (spark.catalog.databaseExists(s\"$databaseName\")) {\n",
    "        println(s\"$databaseName has exists!\")\n",
    "    }else{\n",
    "        spark.sql(s\"create database if not exists $databaseName\").show\n",
    "        spark.sql(s\"use $databaseName\").show\n",
    "        for (table <- tables) {\n",
    "            if (spark.catalog.tableExists(s\"$table\")){\n",
    "                println(s\"$table has exists!\")\n",
    "            }else{\n",
    "                spark.catalog.createTable(s\"$table\", s\"$data_path/$table\", \"arrow\")\n",
    "            }\n",
    "        }\n",
    "        if (partitionTables) {\n",
    "            for (table <- tables) {\n",
    "                try{\n",
    "                    spark.sql(s\"ALTER TABLE $table RECOVER PARTITIONS\").show\n",
    "                }catch{\n",
    "                        case e: Exception => println(e)\n",
    "                }\n",
    "            }\n",
    "        }\n",
    "    }\n",
    "}\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "143c468d-21f2-466b-8bfe-cd3fe997da49",
   "metadata": {},
   "outputs": [],
   "source": [
    "val timeout = 60 // timeout in hours\n",
    "\n",
    "// COMMAND ----------\n",
    "\n",
    "// Spark configuration\n",
    "spark.conf.set(\"spark.sql.broadcastTimeout\", \"10000\") // good idea for Q14, Q88.\n",
    "\n",
    "// ... + any other configuration tuning\n",
    "\n",
    "// COMMAND ----------\n",
    "sql(s\"use $databaseName\")\n",
    "import com.databricks.spark.sql.perf.tpcds.TPCDS\n",
    "val tpcds = new TPCDS (sqlContext = spark.sqlContext)\n",
    "def queries = {\n",
    "  val filtered_queries = query_filter match {\n",
    "    case Seq() => tpcds.tpcds2_4Queries\n",
    "    case _ => tpcds.tpcds2_4Queries.filter(q => query_filter.contains(q.name))\n",
    "  }\n",
    "  if (randomizeQueries) scala.util.Random.shuffle(filtered_queries) else filtered_queries\n",
    "}\n",
    "val experiment = tpcds.runExperiment(\n",
    "  queries, \n",
    "  iterations = iterations,\n",
    "  resultLocation = resultLocation,\n",
    "  tags = Map(\"runtype\" -> \"benchmark\", \"database\" -> databaseName, \"scale_factor\" -> scaleFactor))\n",
    "\n",
    "println(experiment.toString)\n",
    "experiment.waitForFinish(timeout*60*60)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "spylon-kernel",
   "language": "scala",
   "name": "spylon-kernel"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".scala",
   "help_links": [
    {
     "text": "MetaKernel Magics",
     "url": "https://metakernel.readthedocs.io/en/latest/source/README.html"
    }
   ],
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala",
   "version": "0.4.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}