spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "# Sources for this notebook\n",
    "This is based on [this book - Learning Spark](https://learning.oreilly.com/library/view/learning-spark-2nd/9781492050032/). You can register online for orielly, which gives you 10 days of free access.  You can use these days to go go over the book before buying it.\n",
    "\n",
    "Data used in here is available from books [website](https://github.com/databricks/LearningSparkV2.git)\n",
    "\n",
    "Spark API [documentation](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql.html)\n",
    "\n",
    "Spark [Guides](https://spark.apache.org/docs/latest/index.html#where-to-go-from-here)\n",
    "\n",
    "Here is the [sample paper](https://databricks-prod-cloudfront.cloud.databricks.com/public/793177bc53e528530b06c78a4fa0e086/0/6221173/100020/latest.html) from 2018. This is from dataricks website, but is irrelevant today.  In this notebook, I have translated one of the scala question to pyspark.\n",
    "\n",
    "Focus of Databricks exam is dataframe and SQL API. Exam is 80% (or more) is based on code questions. \n",
    "\n",
    "Here is my [badge](https://credentials.databricks.com/28181734-274d-4085-89de-0fcb9b2824f8) and [my spark blogs](https://www.kamalsblog.com/search/label/Spark)"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Setup\n",
    "Install Java (I am on 11), Spark and Python (I have 3.8). \n",
    "\n",
    "Install Jupyter server and jupyter plugin in vscode.\n",
    "\n",
    "Set env variables as below (Change values according to your setup)\n",
    "\n",
    "```\n",
    "export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64\n",
    "export SPARK_HOME=/home/kamal/spark-3.1.2-bin-hadoop3.2\n",
    "export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin\n",
    "export PYTHONPATH=${SPARK_HOME}/python/:$(echo ${SPARK_HOME}/python/lib/py4j-*-src.zip):${PYTHONPATH}\n",
    "```\n"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Create Sessions\n",
    "## Create Spark Session"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = (SparkSession.builder\n",
    ".config(\"spark.app.name\",\"airline\").config(\"spark.sql.shuffle.partitions\", 50).getOrCreate())\n"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Create Delta spark session\n",
    "* This requires delta lake jar be downloaded from mavenrepository and placed in SPARK_HOME/jars. Restart Jupyter server after that.  \n",
    "* Install delta-spark package (pip install delta-spark)"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from delta import *\n",
    "\n",
    "builder = SparkSession.builder.appName(\"MyApp\") \\\n",
    "    .config(\"spark.sql.extensions\", \"io.delta.sql.DeltaSparkSessionExtension\") \\\n",
    "    .config(\"spark.sql.catalog.spark_catalog\", \"org.apache.spark.sql.delta.catalog.DeltaCatalog\")\n",
    "\n",
    "delta_spark = configure_spark_with_delta_pip(builder).getOrCreate()\n"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Read and operate on Data - Departure DS\n",
    "## Departure Delays - Read CSV with string schema\n",
    "Data used here is available from databricks github [page](https://github.com/databricks/LearningSparkV2.git)"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "csv_file = \"./databricks-datasets/learning-spark-v2/flights/departuredelays.csv\"\n",
    "schema = \"date string, delay int, distance int, origin string, destination string\"\n",
    "df = (spark.read\n",
    ".schema(schema)\n",
    ".option(\"header\", \"true\")\n",
    ".csv(csv_file)).cache()\n",
    "df.printSchema()\n",
    "df.first()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Repartition with and without column(s)\n",
    "Repartition can accepts both number and columns while coalease accepts only number.  Coalease cannot increase number of partititions."
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.functions import col\n",
    "from pyspark.conf import SparkConf\n",
    "conf = SparkConf()\n",
    "print(spark.sparkContext.getConf().getAll())\n",
    "print(\"Original number of partitions are {}\".format(df.rdd.getNumPartitions()))\n",
    "print(\"Number of partitions after repartiton are {}\".format(df.repartition(8).rdd.getNumPartitions()))\n",
    "# This will max out at the limit set when the session was created (very first cell)\n",
    "print(\"Number of partitions after repartiton with column 'origin' are {}\".format(df.repartition(\"origin\").rdd.getNumPartitions()))\n",
    "print(\"Number of partitions after repartiton with column are {}\".format(df.repartition(\"origin\", \"destination\").rdd.getNumPartitions()))\n",
    "\n",
    "print(\"Number of partitions after coalease do not increase -- {}\".format(df.repartition(8).coalesce(16).rdd.getNumPartitions()))\n",
    "print(\"Number of partitions after coalease decrease -- {}\".format(df.repartition(8).coalesce(4).rdd.getNumPartitions()))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Operate on Columns, with col"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.functions import *\n",
    "try:\n",
    "    df.select(col(\"distance\") / 1000).show(2)\n",
    "except:\n",
    "    print(\"Operation with col failed\")\n",
    "\n",
    "try:\n",
    "    df.select(\"distance\" / 1000).show(2)\n",
    "except:\n",
    "    print(\"Operation without col failed\")\n",
    "\n",
    "try:\n",
    "    df.select(\"distance\").show(2)\n",
    "except:\n",
    "    print(\"select with just quotes failed\")\n"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Using withColumn() to add new computed column"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df.withColumn(\"calculatedCol\", col(\"distance\")/1000).show(2)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Difference between distinct and drop_duplicate \n",
    "Drop duplicate accepts columns while distinct does not"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "print(df.select(\"origin\").distinct().count())\n",
    "print(df.select(\"origin\", \"destination\").distinct().count())\n",
    "print(df.drop_duplicates([\"origin\",\"destination\"]).count())\n",
    "try:\n",
    "    print(df.drop_duplicates([col(\"origin\"),col(\"destination\")]).count())\n",
    "except:\n",
    "    print(\"dropDuplicates does not work with col function\")    "
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Timestamp from millisecond (since epoch)"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"select timestamp_millis(1631163333673) as timestampdata\").show(truncate=False)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Select and GroupBy"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# df.printSchema()\n",
    "from pyspark.sql.functions import *\n",
    "df_agg = df.select(col(\"origin\"), \"destination\",\"delay\").groupby(\"origin\", \"destination\")\n",
    "# Works with or without col or with dictionary\n",
    "# Cannot mix dictioanry with columns\n",
    "df_agg.agg(min(col(\"delay\")), max(\"delay\")).show(5)\n",
    "df_agg.agg({\"delay\" : \"min\", \"delay\": \"max\"}).show(5)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Working with JSON\n",
    "\n",
    "### Extract with from_json()"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "data = [(1, '''{\"a\": 1,\"b\": 2.3}''')]\n",
    "schema = 'a INT, b FLOAT'\n",
    "df_json = spark.createDataFrame(data, (\"key\", \"value\"))\n",
    "df_json.select(from_json(\"value\", schema)[\"a\"].alias(\"id\"), \n",
    "from_json(df_json.value, schema)[\"b\"].alias(\"value\")).show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Extract with get_json_object()"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df_json.select(get_json_object(df_json.value,\"$.a\").alias(\"id\"), \n",
    "get_json_object(\"value\",\"$.b\").alias(\"value\")).show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Filter Dataframe"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df.printSchema()\n",
    "from pyspark.sql.functions import col\n",
    "try:\n",
    "    df.select(\"distance\",\"origin\",\"destination\").filter(\"distance > 1000\").distinct().show(2)\n",
    "except:\n",
    "    print(\"Filter without col function works with SQL like clause does not work\")\n",
    "\n",
    "try:\n",
    "    df.select(\"distance\",\"origin\",\"destination\").filter(\"distance\" > 1000).distinct().show(2)\n",
    "except:\n",
    "    print(\"Filter without col with only string quotes does not work\")\n",
    "\n",
    "try:\n",
    "    df.select(\"distance\",\"origin\",\"destination\").filter(col(\"distance\") > 1000).distinct().show(2)\n",
    "except:\n",
    "    print(\"Filter with col with python syntax does not work\")\n"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Create Views\n",
    "* Global views are used, in queries, with \"global_temp\" prefix."
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df.createOrReplaceTempView(\"Airline\")\n",
    "df.createOrReplaceGlobalTempView(\"GlobalAirlineView\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"select * from Airline limit 2\").show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"\"\"SELECT delay, origin, destination, \n",
    "              CASE\n",
    "                  WHEN delay > 360 THEN 'Very Long Delays'\n",
    "                  WHEN delay >= 120 AND delay <= 360 THEN 'Long Delays'\n",
    "                  WHEN delay >= 60 AND delay < 120 THEN 'Short Delays'\n",
    "                  WHEN delay > 0 and delay < 60 THEN 'Tolerable Delays'\n",
    "                  WHEN delay = 0 THEN 'No Delays'\n",
    "                  ELSE 'Early'\n",
    "               END AS Flight_Delays\n",
    "               FROM Airline\n",
    "               ORDER BY origin, delay DESC\"\"\").show(10)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"\"\"SELECT *,\n",
    "                CASE\n",
    "                    WHEN delay < 0 THEN 'EARLY'\n",
    "                    WHEN delay == 0 THEN 'ON-TIME'\n",
    "                    ELSE 'DELAY'\n",
    "                END AS FL_Delay\n",
    "             FROM Airline\"\"\").show(6)   "
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Add column to DF\n",
    "### Use multiple conditions on column with when-otherwise"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.functions import *\n",
    "df.withColumn('FL_delay', when(col(\"delay\") > 0, 'DELAYED')\n",
    ".when(col(\"delay\") < 0, 'EARLY')\n",
    ".otherwise('ON-TIME')).show(6)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Set shuffle partitions"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.conf.set(\"spark.sql.shuffle.partitions\",4)\n",
    "df.write.option(\"path\",\"/home/kamal/airline_data\").format(\"csv\").saveAsTable(\"Airline\")\n",
    "df.rdd.getNumPartitions()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Create, Use and drop databases"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.catalog.listDatabases()\n",
    "spark.sql(\"drop database if exists newdb cascade\")\n",
    "spark.sql(\"create database newdb location '/home/kamal/spark-warehouse/newdatabase'\")\n",
    "spark.catalog.listDatabases()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"use newdb\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df.write.saveAsTable(\"airline_date\", mode=\"overwrite\", format=\"parquet\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"select * from airline_date where origin = 'SFO'\").createOrReplaceGlobalTempView(\"us_origin_airport_SFO_global\")\n",
    "spark.sql(\"select * from global_temp.us_origin_airport_SFO_global\").count()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"show views\").show()\n",
    "spark.catalog.dropGlobalTempView(\"us_origin_airport_SFO_global\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Working with AVRO\n",
    "Avro package is still external (but supported), so needs to be downloaded and placed in classpath. Download spark-avro jar from [mavenrepo](https://mvnrepository.com/artifact/org.apache.spark/spark-avro_2.12/3.1.2),  and place it in $SPARK_HOME/jars. Then restart Jupyter server "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.conf.set(\"spark.jars.packages\", \"com.databricks:spark-avro_2.12:3.1.2\")\n",
    "spark.sql(\"\"\"\n",
    "CREATE OR REPLACE TEMPORARY VIEW episode_tbl\n",
    "    USING com.databricks.spark.avro\n",
    "    OPTIONS (\n",
    "      path \"./databricks-datasets/learning-spark-v2/flights/summary-data/avro/*\"\n",
    "    )\n",
    "\"\"\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df_fl_summary = spark.sql(\"select * from episode_tbl\")\n",
    "df_fl_summary.count()\n",
    "df_fl_summary.show(5)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# UDFs"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.types import LongType\n",
    "import pandas as pd\n",
    "def cubed(s: pd.Series) -> pd.Series:\n",
    "    return s*s*s\n",
    "\n",
    "df_range = spark.range(1,9)    \n",
    "cubed_udf = pandas_udf(cubed, returnType=LongType())"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df_range.select(\"id\",cubed_udf(\"id\").alias(\"cubed\")).show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.udf.register(\"cubed_udf\",cubed_udf)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df_range.createOrReplaceTempView(\"udf_test\")\n",
    "spark.sql(\"select id, cubed_udf(id) from udf_test\").show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Delta Read and Write"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "data = delta_spark.sql(\"select * from global_temp.GlobalAirlineView\").cache()\n",
    "data.write.format(\"delta\").option(\"mergeSchema\", \"true\").mode(\"overwrite\").save(\"/tmp/delta-table/\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df = spark.read.format(\"delta\").load(\"/tmp/delta-table\")\n",
    "df.show(5)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# NFL Data - Working with NA\n",
    "\n",
    "Data is available [here](https://www.kaggle.com/rtatman/data-cleaning-challenge-handling-missing-values)\n",
    "\n",
    "In this CSV I have replaced all NA values with blanks. \n",
    "\n",
    "## Load data"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.types import NullType\n",
    "nfl_df = (spark.read.options(header = True)\n",
    ".options(emptyValue = NullType)\n",
    ".options(nullValue = NullType)\n",
    ".csv(\"nfldata.csv\"))\n",
    "nfl_df.limit(2).toPandas()\n"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## dropna\n",
    "In the function definition (see apsrk documentation), default for \"how\" is 'any'\n",
    "\n",
    "Threshold field in this function decides how many missing columns in a row are allowed"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "print(\"Number of Columns: \",len(nfl_df.columns))\n",
    "total_rows = nfl_df.count()\n",
    "print(\"Number of Rows: \",total_rows)\n",
    "clean_rows = nfl_df.dropna().count()\n",
    "na_rows = total_rows - clean_rows\n",
    "print(\"Rows where atleast one column is missing: \",na_rows)\n",
    "clean_rows_with_thresh_101 = nfl_df.dropna(thresh=101).count()\n",
    "print(\"Rows where more than one column is missing: \",total_rows - clean_rows_with_thresh_101)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.functions import *\n",
    "nfl_df_101 = nfl_df.dropna(thresh=101)\n",
    "nfl_rows_with_na = nfl_df.exceptAll(nfl_df_101)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "nfl_rows_with_na.limit(5).toPandas()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# This means our dataset will not have any row where either \"Win_Prod\" is null or \"airWPA\" is null\n",
    "nfl_both_cols_ok = nfl_df.dropna(how=\"all\",subset = [\"Win_Prob\",\"airWPA\"])\n",
    "nfl_both_cols_not_ok = nfl_df.exceptAll(nfl_both_cols_ok)\n",
    "nfl_both_cols_not_ok.limit(5).toPandas()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Spark Streaming\n",
    "Spark streaming is not part of Databricks Spark exam. Do check syllabus time to time\n",
    "## foreachBatch Function\n",
    "\n",
    "foreachBatch lets us use \"update\", \"append\" and \"complete\" output modes. Whereas file sink only allows append."
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from functools import partial\n",
    "def foreachBatchFunction(format, location, df, batchId):\n",
    "        (df\n",
    "        .write\n",
    "        .mode(\"overwrite\")\n",
    "        .format(format)\n",
    "        .save(location+str(batchId)))\n",
    "writeMicroBatchUpdateToJSON = partial(foreachBatchFunction,\"JSON\",\"/home/kamal/jsonUpdate/\")\n",
    "writeMicroBatchAppendToJSON = partial(foreachBatchFunction,\"JSON\",\"/home/kamal/jsonAppend/\")\n",
    "writeMicroBatchCompleteToJSON = partial(foreachBatchFunction,\"JSON\",\"/home/kamal/jsonComplete/\")\n",
    "writeMicroBatchAppendToJSON10Sec = partial(foreachBatchFunction,\"JSON\",\"/home/kamal/jsonAppend10sec/\")\n",
    "writeMicroBatchToCSV = partial(foreachBatchFunction,\"JSON\",\"/home/kamal/sparkcsv/\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Word Count\n",
    "Generate test dat using [datastreamer](https://github.com/skamalj/datagenerator) - with followung config. It creates space separated list of  6 words in each line\n",
    ">`lorem|words|textline|6`"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import *\n",
    "sparkstream = (SparkSession.builder\n",
    ".appName(\"airline\").getOrCreate())\n",
    "lines = (sparkstream\n",
    "  .readStream.format(\"socket\")\n",
    "  .option(\"host\", \"localhost\")\n",
    "  .option(\"port\", 4000)\n",
    "  .load())\n",
    "\n",
    "linesnojson = lines.select(get_json_object(\"value\",\"$.textline\").alias(\"line\"))\n",
    "\n",
    "words = linesnojson.select(explode(split(col(\"line\"), \" \")).alias(\"word\"))\n",
    "counts = words.groupBy(\"word\").count()\n",
    "\n",
    "checkpointDir = \"/home/kamal/sparkcheckpoint\"\n",
    "checkpointFileSyncDir = \"/home/kamal/sparkcheckpointcsv\"\n",
    "\n",
    "#streamingQuery = (counts\n",
    "#  .writeStream\n",
    "#  .format(\"console\")\n",
    "#  .outputMode(\"complete\")\n",
    "#  .trigger(processingTime=\"20 second\")\n",
    "#  .option(\"checkpointLocation\", checkpointDir)\n",
    "#  .start())\n",
    "\n",
    "streamingQueryFile = (counts\n",
    "  .writeStream\n",
    "  .foreachBatch(writeMicroBatchToCSV)\n",
    "  .outputMode(\"update\")\n",
    "  .trigger(processingTime=\"20 second\")\n",
    "  .option(\"checkpointLocation\", checkpointFileSyncDir)\n",
    "  .start())"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "#streamingQuery.stop()\n",
    "streamingQueryFile.stop()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Streaming with eventtime\n",
    "Use generator with following config\n",
    ">`datatype|number|id|{\"min\":0,\"max\":10}`\n",
    "\n",
    ">`datatype|float|temp|{\"min\":20,\"max\":50}`"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import *\n",
    "sparkstream = (SparkSession.builder\n",
    ".appName(\"avgtemp\").getOrCreate())\n",
    "readings = (sparkstream\n",
    "  .readStream.format(\"socket\")\n",
    "  .option(\"host\", \"localhost\")\n",
    "  .option(\"port\", 4000)\n",
    "  .load().withColumn(\"current_timestamp\", current_timestamp()))\n",
    "\n",
    "readings_tab = readings.select(get_json_object(\"value\",\"$.id\").alias(\"id\"),\n",
    "get_json_object(\"value\",\"$.temp\").alias(\"temp\"),\n",
    "timestamp_seconds(round(get_json_object(\"value\",\"$.eventtime\") / 1000)).alias(\"eventtime\"),\n",
    "\"current_timestamp\")\n",
    "\n",
    "readings_tab = readings_tab.withColumn(\"delayed_eventtime\", col(\"eventtime\") - expr(\"INTERVAL 10 seconds\"))\n",
    "\n",
    "eventcol = \"delayed_eventtime\"\n",
    "\n",
    "samples = (readings_tab.withWatermark(eventcol, \"10 seconds\")\n",
    ".groupBy(\"id\", window(eventcol, \"20 second\"))\n",
    ".agg(count(\"id\").alias(\"samples\"),avg(\"temp\").alias(\"averageTemp\"),\n",
    "max(\"current_timestamp\").alias(\"processed_at\"), max(eventcol).alias(\"generated_at\")))\n",
    "\n",
    "## Cooment / uncomment below blocks as per need\n",
    "\n",
    "#streamingQuery = (samples\n",
    "#  .writeStream\n",
    "#  .format(\"console\")\n",
    "#  .option(\"truncate\", \"false\")\n",
    "#  .outputMode(\"append\")\n",
    "#  .trigger(processingTime=\"10 second\")\n",
    "#  .start())\n",
    "#\n",
    "#streamingQueryJsonAppend = (samples\n",
    "#  .writeStream\n",
    "#  .foreachBatch(writeMicroBatchAppendToJSON10Sec)\n",
    "#  .outputMode(\"append\")\n",
    "#  .option(\"checkpointLocation\", \"/home/kamal/spark/checkpoint/jsonappend10sec\")\n",
    "#  .trigger(processingTime=\"10 second\")\n",
    "#  .start())   \n",
    "\n",
    "streamingQueryJsonAppend = (samples\n",
    "  .writeStream\n",
    "  .foreachBatch(writeMicroBatchAppendToJSON)\n",
    "  .outputMode(\"append\")\n",
    "  .option(\"checkpointLocation\", \"/home/kamal/spark/checkpoint/jsonappend\")\n",
    "  .trigger(processingTime=\"20 second\")\n",
    "  .start())   \n",
    "\n",
    "#streamingQueryJsonUpdate = (samples\n",
    "#  .writeStream\n",
    "#  .foreachBatch(writeMicroBatchUpdateToJSON)\n",
    "#  .outputMode(\"update\")\n",
    "#  .option(\"checkpointLocation\", \"/home/kamal/spark/checkpoint/jsonupdate\")\n",
    "#  .trigger(processingTime=\"20 second\")\n",
    "#  .start())   \n",
    "\n",
    "#streamingQueryJsonComplete = (samples\n",
    "#  .writeStream\n",
    "#  .foreachBatch(writeMicroBatchCompleteToJSON)\n",
    "#  .outputMode(\"complete\")\n",
    "#  .option(\"checkpointLocation\", \"/home/kamal/spark/checkpoint/jsoncomplete\")\n",
    "#  .start())"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Stop Streaming Queries"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "#streamingQueryJsonUpdate.stop()\n",
    "streamingQueryJsonAppend.stop()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "dfc = spark.read.format(\"delta\").load(\"/tmp/deltaeventscomplete\").cache()\n",
    "dfc.filter(\"id == 2\").show(15, truncate = False)\n"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "df = spark.read.format(\"delta\").load(\"/tmp/deltaevents\").cache()\n",
    "df.filter(\"id == 2\").show(15, truncate = False)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "\n",
    "from pyspark.sql.functions import col\n",
    "(df.join(dfc,['id','window'], \"fullOuter\")\n",
    "    .select(\"id\", \"window\", dfc.samples.alias(\"complete\"), df.samples,dfc.processed_at)\n",
    "    .orderBy(\"id\")\n",
    "    .filter(\"id == 10\")\n",
    "    .show(50, truncate=False))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# People DS - Scala excercise from [Sample question](https://databricks-prod-cloudfront.cloud.databricks.com/public/793177bc53e528530b06c78a4fa0e086/0/6221173/100020/latest.html). Translated to Pyspark"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## People  - Imports and Create DF"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "#schema = StructType([\n",
    "#  StructField(\"name\", StringType(), True),\n",
    "#  StructField(\"department\", IntegerType(), True),\n",
    "#  StructField(\"score\", ArrayType(IntegerType()), True)\n",
    "#])\n",
    "\n",
    "# Works same as above\n",
    "schema = \"name string, department string, score array<int>\"\n",
    "\n",
    "peopleDF = spark.createDataFrame([\n",
    "  (\"Ali\", 0, [100]),\n",
    "  (\"Barbara\", 1, [300, 250, 100]),\n",
    "  (\"Cesar\", 1, [350, 100]),\n",
    "  (\"Dongmei\", 1, [400, 100]),\n",
    "  (\"Eli\", 2, [250]),\n",
    "  (\"Florita\", 2, [500, 300, 100]),\n",
    "  (\"Gatimu\", 3, [300, 100])\n",
    "],schema) "
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "peopleDF.schema"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Create windowSpec"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.window import Window\n",
    "\n",
    "windowSpec = Window.partitionBy(\"department\").orderBy(col(\"escore\").desc())\n"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Execute using windowSpec"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "people = (peopleDF.select(\"*\", explode(\"score\")\n",
    ".alias(\"escore\"))\n",
    ".drop(\"score\")\n",
    ".select(\"name\", expr(\"department *1.0\").alias(\"Dept\"),\"escore\", rank().over(windowSpec).alias(\"maxrank\"))\n",
    ".filter(\"maxrank == 1\")\n",
    ".drop(\"maxrank\")\n",
    ".orderBy(\"department\", ascending=False)\n",
    ".show(5))\n"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Explode and collect_list"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "people_listscore = peopleDF.select(\"name\",\"department\",explode(\"score\").alias(\"score\"))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "people_listscore.groupBy(\"name\").agg(collect_list(\"score\")).show(20)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "mnmdata = \"/home/kamal/Documents/dev/spark/LearningSparkV2/chapter2/py/src/data/mnm_dataset.csv\"\n",
    "\n",
    "mnmDF = (spark\n",
    ".read\n",
    ".format(\"csv\")\n",
    ".option(\"header\",\"True\")\n",
    ".option(\"inferschema\", \"True\")\n",
    ".load(mnmdata))\n",
    "\n",
    "resultDF = (mnmDF\n",
    ".groupBy(\"State\",\"Color\")\n",
    ".agg(sum(\"count\").alias(\"Total\"))\n",
    ".orderBy(\"Total\",ascending = False)\n",
    ")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "resultDF.count()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Show current spark context, get the UI url"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sparkContext"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Working with string schema"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "data = [[1, \"Jules\", \"Damji\", \"https://tinyurl.1\", \"1/4/2016\", 4535, [\"twitter\",\n",
    "\"LinkedIn\"]],\n",
    "       [2, \"Brooke\",\"Wenig\", \"https://tinyurl.2\", \"5/5/2018\", 8908, [\"twitter\",\n",
    "\"LinkedIn\"]],\n",
    "       [3, \"Denny\", \"Lee\", \"https://tinyurl.3\", \"6/7/2019\", 7659, [\"web\",\n",
    "\"twitter\", \"FB\", \"LinkedIn\"]],\n",
    "       [4, \"Tathagata\", \"Das\", \"https://tinyurl.4\", \"5/12/2018\", 10568, \n",
    "[\"twitter\", \"FB\"]],\n",
    "       [5, \"Matei\",\"Zaharia\", \"https://tinyurl.5\", \"5/14/2014\", 40578, [\"web\",\n",
    "\"twitter\", \"FB\", \"LinkedIn\"]],\n",
    "       [6, \"Reynold\", \"Xin\", \"https://tinyurl.6\", \"3/2/2015\", 25568, \n",
    "[\"twitter\", \"LinkedIn\"]]\n",
    "      ]\n",
    "\n",
    "schema = \"\"\"`Id` INT, `First` STRING, `Last` STRING, `Url` STRING, \\\n",
    "`Published` STRING, `Hits` INT, `Campaigns` ARRAY<STRING>\"\"\""
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "blogs = spark.createDataFrame(data,schema)\n",
    "#blogs.write.format(\"json\").save(\"blogs-json\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "blogs.withColumn(\"BigHitters\", expr(\"Hits > 10000\")).filter(col(\"First\").endswith(\"i\")).show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Fire Department Dataset excercise"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Load Dataset"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "firecallsDF = (spark\n",
    ".read\n",
    ".format(\"csv\")\n",
    ".option(\"samplingRatio\", .001)\n",
    ".option(\"header\", True)\n",
    ".load(\"databricks-datasets/learning-spark-v2/sf-fire/sf-fire-calls.csv\"))\n",
    "\n",
    "firecallsDF.select(\"CallType\").distinct().show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### String to Timestamp"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.functions import *\n",
    "new_fire_df = (firecallsDF\n",
    ".select(\"*\",\n",
    "    to_timestamp(col(\"CallDate\"),\"MM/dd/yyyy\").alias(\"IncidentDate\"),\n",
    "    to_timestamp(col(\"WatchDate\"), \"MM/dd/yyyy\").alias(\"OnWatchDate\"),\n",
    "    to_timestamp(col(\"AvailableDtTm\"),\"MM/dd/yyyy hh:mm:ss a\").alias(\"AvailableDtTs\"))\n",
    ".where(col(\"Incidentdate\").isNotNull())\n",
    ".drop(\"CallDate\")\n",
    ".drop(\"WatchDate\")\n",
    ".drop(\"AvailableDtTm\"))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Which calltype generated most calls"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "(firecallsDF\n",
    ".select(\"CallType\")\n",
    ".filter(col(\"CallType\").isNotNull())\n",
    ".groupBy(\"CallType\")\n",
    ".agg(count(\"CallType\").alias(\"count\"))\n",
    ".orderBy(\"count\", ascending=False)\n",
    ".show(30, truncate = False))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### What were all the different types of fire calls in 2018?"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "(new_fire_df\n",
    ".select(\"CallType\",year(\"IncidentDate\").alias(\"Year\"))\n",
    ".filter(\"Year == 2018\")\n",
    ".groupBy(\"CallType\")\n",
    ".agg(count(\"CallType\").alias(\"count\"))\n",
    ".show(2)\n",
    ")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "new_fire_df.columns"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Which neighborhood in San Francisco generated the most fire calls in 2018"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "(new_fire_df\n",
    ".select(\"city\",\"Neighborhood\",\"CallType\",year(\"IncidentDate\").alias(\"Year\"))\n",
    ".where(\"city in ('San Francisco','SF', 'SAN FRANCISCO')\")\n",
    ".where(col(\"Year\") == 2018)  #Can also be written as \"Year == 2018\"\n",
    ".where(col(\"CallType\").like('%Fire%'))\n",
    ".groupBy(\"Neighborhood\")\n",
    ".agg(count(\"Neighborhood\").alias(\"TotalCalls\"))\n",
    ".orderBy(\"TotalCalls\",ascending=False)\n",
    ".show(3, truncate=False))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# UDF vs Pandas UDF"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import LongType\n",
    "\n",
    "def cube(s):\n",
    "    return s*s*s\n",
    "\n",
    "spark.udf.register(\"cube\", cube, LongType())\n",
    "\n",
    "spark.range(1,5).createOrReplaceTempView(\"udf_test\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"select id, cube(id) from udf_test\").show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "cubed_udf = pandas_udf(cube, LongType())\n",
    "df_id = spark.range(1,4)\n",
    "df_id.select(\"id\",cubed_udf(\"id\")).show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from pyspark.sql.functions import expr\n",
    "tripdelaysFilePath = \"databricks-datasets/learning-spark-v2/flights/departuredelays.csv\"\n",
    "airportsnaFilePath =  \"databricks-datasets/learning-spark-v2/flights/airport-codes-na.txt\"\n",
    "  \n",
    "# Obtain airports data set\n",
    "airportsna = (spark.read\n",
    "  .format(\"csv\")\n",
    "  .options(header=\"true\", inferSchema=\"true\", sep=\"\\t\")\n",
    "  .load(airportsnaFilePath))\n",
    "\n",
    "airportsna.createOrReplaceTempView(\"airports_na\")\n",
    "\n",
    "departureDelays = (spark.read\n",
    "  .format(\"csv\")\n",
    "  .options(header=\"true\")\n",
    "  .load(tripdelaysFilePath))\n",
    "\n",
    "departureDelays = (departureDelays\n",
    ".withColumn(\"delay\", expr(\"CAST(delay as INT) as delay\"))\n",
    ".withColumn(\"distance\", expr(\"CAST(distance as INT) as distance\")))\n",
    "\n",
    "departureDelays.createOrReplaceTempView(\"departureDelays\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# SQL Collection functions"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "peopleDF.createOrReplaceGlobalTempView(\"peopleDF\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"select filter(score, s -> s = 250) as new_score from global_temp.peopleDF\").show(truncate=False)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "spark.sql(\"set -v\").select(\"key\",\"value\").show(truncate=False)"
   ],
   "outputs": [],
   "metadata": {}
  }
 ],
 "metadata": {
  "orig_nbformat": 4,
  "language_info": {
   "name": "python",
   "version": "3.8.10",
   "mimetype": "text/x-python",
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "pygments_lexer": "ipython3",
   "nbconvert_exporter": "python",
   "file_extension": ".py"
  },
  "kernelspec": {
   "name": "python3",
   "display_name": "Python 3.8.10 64-bit"
  },
  "interpreter": {
   "hash": "31f2aee4e71d21fbe5cf8b01ff0e069b9275f58929596ceb00d14d90e3e16cd6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}