03 - Streaming Aggregations - Databricks.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f2dc4f11-0f5b-456b-a68f-14eeb4043b55",
     "showTitle": false,
     "title": ""
    },
    "id": "ICXbyhtyjXZM"
   },
   "source": [
    "#**Spark Structured Streaming: Aggregations**\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ad2f815d-d5ea-4350-99b3-226bcdf55ccd",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "de163f53-d506-4291-9700-21428a0c1eb8",
     "showTitle": false,
     "title": ""
    },
    "id": "dvt7tK1CBqK1"
   },
   "source": [
    "# Data: Heterogeneity Human Activity Recognition Dataset\n",
    "* In this notebook, we will work with the [Heterogeneity Human Activity Recognition Dataset](https://archive.ics.uci.edu/dataset/344/heterogeneity+activity+recognition).\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "fe62f1fe-08d5-4f9f-b551-188aab5779f7",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%fs ls /databricks-datasets/definitive-guide/data/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "cc7b1ef6-0a2a-4ed6-b5e0-17407b744115",
     "showTitle": false,
     "title": ""
    },
    "id": "Q6lcZ7rw49Rm"
   },
   "source": [
    "## Create a Static DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "91f1f7e0-3558-42f2-a7d4-b4e785f02ca2",
     "showTitle": false,
     "title": ""
    },
    "id": "oP86sYkICzbu"
   },
   "outputs": [],
   "source": [
    "sensorDataPath = \"/databricks-datasets/definitive-guide/data/activity-data\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d2ad4509-0b19-4c4f-9f59-167dde007363",
     "showTitle": false,
     "title": ""
    },
    "id": "vPHnRFcOCcwn"
   },
   "outputs": [],
   "source": [
    "spark.conf.set(\"spark.sql.shuffle.partitions\", 5)\n",
    "\n",
    "sensorStatic = spark.read.json(sensorDataPath)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "201c9a93-07d0-4732-8c03-8acfc87e420e",
     "showTitle": false,
     "title": ""
    },
    "id": "133D6oUvESt_"
   },
   "outputs": [],
   "source": [
    "sensorStatic.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "842d7965-5c4e-4075-9585-0f4525725985",
     "showTitle": false,
     "title": ""
    },
    "id": "TlLCoAUpjFaY"
   },
   "outputs": [],
   "source": [
    "sensorStatic.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5e3c6035-8c2f-4fb9-a3a6-478bbd711d20",
     "showTitle": false,
     "title": ""
    },
    "id": "T8gwKgfBjQp4"
   },
   "source": [
    "## Create a Streaming DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "39ca49ac-6672-4cd8-bbf8-3f4976f068cc",
     "showTitle": false,
     "title": ""
    },
    "id": "aW6QOSjTEFj3"
   },
   "outputs": [],
   "source": [
    "sensorStreaming = spark\\\n",
    "  .readStream\\\n",
    "  .schema(sensorStatic.schema)\\\n",
    "  .option(\"maxFilesPerTrigger\", 10)\\\n",
    "  .json(sensorDataPath)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "7026a045-aa4e-439a-8590-0f7992ebdec2",
     "showTitle": false,
     "title": ""
    },
    "id": "w3rnmRROE2xN"
   },
   "source": [
    "# Streaming Aggregations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "61c90751-f135-4d68-bf16-267f704f98b9",
     "showTitle": false,
     "title": ""
    },
    "id": "38LZNWKQE9qO"
   },
   "source": [
    "## Aggregations not based on time\n",
    "* Aggregations not involving time can be broadly classified into two categories: global aggregations and grouped aggregations."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "14b7ec8b-4563-4228-af30-51d92bf52aa6",
     "showTitle": false,
     "title": ""
    },
    "id": "hM91oYYhkTbv"
   },
   "source": [
    "### Global aggregations\n",
    "* Aggregations across all the data in the stream."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1ad12372-2dd2-42d3-8ee0-4de7309eff98",
     "showTitle": false,
     "title": ""
    },
    "id": "T1uYD3qSkZrv"
   },
   "outputs": [],
   "source": [
    "globalCountQuery = sensorStreaming.groupBy().count()\\\n",
    " .writeStream\\\n",
    " .queryName(\"globalCount\")\\\n",
    " .format(\"memory\")\\\n",
    " .outputMode(\"complete\")\\\n",
    " .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e7687130-49a2-49f6-9911-03783d431076",
     "showTitle": false,
     "title": ""
    },
    "id": "TLUKxwHpolys"
   },
   "outputs": [],
   "source": [
    "spark.table(\"globalCount\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9aa80c99-d5ad-408c-b50a-f171e11cda5e",
     "showTitle": false,
     "title": ""
    },
    "id": "Q5-68K1AkaTv"
   },
   "source": [
    "### Grouped aggregations\n",
    "* Aggregations within each group or key present in the data stream."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "0e369e59-8b2a-4923-bcb7-53a2f7c3c5e3",
     "showTitle": false,
     "title": ""
    },
    "id": "1dP2uWWlkgLS"
   },
   "outputs": [],
   "source": [
    "groupedCountQuery = sensorStreaming.groupBy(\"gt\").count()\\\n",
    " .writeStream\\\n",
    " .queryName(\"gtCount\")\\\n",
    " .format(\"memory\")\\\n",
    " .outputMode(\"complete\")\\\n",
    " .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e5bbabce-cc5f-4153-8717-46e19771d114",
     "showTitle": false,
     "title": ""
    },
    "id": "3gAoMBj-pcO0"
   },
   "outputs": [],
   "source": [
    "spark.table(\"gtCount\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d16e5b1a-5602-45ed-97e4-f3b81c0bd9a9",
     "showTitle": false,
     "title": ""
    },
    "id": "vKKla425lOa_"
   },
   "source": [
    "### Multiple aggregations computed together"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ea8811bf-3374-44f4-b309-6b68815b5a33",
     "showTitle": false,
     "title": ""
    },
    "id": "rtUjh_bIlPlm"
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *\n",
    "\n",
    "multipleAggQuery = sensorStreaming.groupBy(\"gt\")\\\n",
    "  .agg(count(\"*\"), approx_count_distinct(\"Device\").alias(\"deviceCount\"),approx_count_distinct(\"Model\").alias(\"modelCount\"))\\\n",
    "  .writeStream\\\n",
    "  .queryName(\"multipleAggs\")\\\n",
    "  .format(\"memory\")\\\n",
    "  .outputMode(\"complete\")\\\n",
    "  .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "30c1f654-743b-48e2-a2cb-76d88490a9e3",
     "showTitle": false,
     "title": ""
    },
    "id": "zcX0Doh1qYxz"
   },
   "outputs": [],
   "source": [
    "spark.table(\"multipleAggs\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "24d936b1-ddac-4c0c-b10e-86c8ae14893b",
     "showTitle": false,
     "title": ""
    },
    "id": "q22f05WftQh8"
   },
   "outputs": [],
   "source": [
    "#it is recommneded to stop previous queries before running the following ones\n",
    "globalCountQuery.stop()\n",
    "groupedCountQuery.stop()\n",
    "multipleAggQuery.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1c6b161c-5cb1-497b-9714-9b8f112460b5",
     "showTitle": false,
     "title": ""
    },
    "id": "yh3ZBx11FBFn"
   },
   "source": [
    "## Aggregations with Event-Time Windows\n",
    "* Spark supports [three types of time windows](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#types-of-time-windows): tumbling, sliding and session.\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b2411470-3e75-4312-8543-56014ad34786",
     "showTitle": false,
     "title": ""
    },
    "id": "GFhptZGHD963"
   },
   "source": [
    "### Event Time Column\n",
    "* In this dataset, there are two time-based columns. The *Creation_Time* column defines when an event was created at the source sensor, whereas the *Arrival_Time* defines when an event hits the servers.\n",
    "* We will use *Creation_Time* as the event time in this example.\n",
    "* The first step in event-time processing is to convert the event time column into the proper Spark SQL timestamp type.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c9f1a2ee-c458-4c48-89ba-82c8b4e692d3",
     "showTitle": false,
     "title": ""
    },
    "id": "2ytbW1XFDhTX"
   },
   "outputs": [],
   "source": [
    "sensorStreamingEvent = sensorStreaming.selectExpr(\n",
    "  \"*\",\n",
    "  \"cast(cast(Creation_Time as double)/1000000000 as timestamp) as event_time\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "0c3b644c-37cc-4f51-a22c-10b6505c4266",
     "showTitle": false,
     "title": ""
    },
    "id": "wPgBnp_gEYwG"
   },
   "outputs": [],
   "source": [
    "sensorStreamingEvent.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e070d461-a32d-4b26-8d67-f08a6ad0af1b",
     "showTitle": false,
     "title": ""
    },
    "id": "k7PCsTXiFT_l"
   },
   "source": [
    "### Tumbling Windows\n",
    "* Tumbling windows are a series of fixed-sized, non-overlapping and contiguous time intervals.\n",
    "* An input can only be bound to a single window.\n",
    "* Tumbling windows use the window function: [pyspark.sql.functions.window](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.window.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "df51576d-5de4-4a6f-9c5e-ded3a6475e3c",
     "showTitle": false,
     "title": ""
    },
    "id": "ctolRFs2EvD9"
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import window, col\n",
    "\n",
    "#ten-minute count\n",
    "tumblingWindowQuery1 = sensorStreamingEvent.groupBy(window(col(\"event_time\"), windowDuration=\"10 minutes\")).count()\\\n",
    "  .writeStream\\\n",
    "  .queryName(\"events_per_window1\")\\\n",
    "  .format(\"memory\")\\\n",
    "  .outputMode(\"complete\")\\\n",
    "  .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "dcc59291-8e24-454a-be61-d646f5537acc",
     "showTitle": false,
     "title": ""
    },
    "id": "nLQxq5AFE0yu"
   },
   "outputs": [],
   "source": [
    "spark.table(\"events_per_window1\").printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "29a26f1f-c4a5-436f-90ac-079c7367ba36",
     "showTitle": false,
     "title": ""
    },
    "id": "reBh0xDwGZlE"
   },
   "outputs": [],
   "source": [
    "spark.table(\"events_per_window1\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "dfae4008-b10c-4951-a628-6d3a1d8ec14c",
     "showTitle": false,
     "title": ""
    },
    "id": "0n0VRuvlFit5"
   },
   "outputs": [],
   "source": [
    "#aggregation on multiple columns including event time\n",
    "tumblingWindowQuery2 = sensorStreamingEvent.groupBy(window(col(\"event_time\"), \"10 minutes\"), \"User\").count()\\\n",
    "  .writeStream\\\n",
    "  .queryName(\"events_per_window2\")\\\n",
    "  .format(\"memory\")\\\n",
    "  .outputMode(\"complete\")\\\n",
    "  .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "3094a1b4-e572-4d71-afb5-7d7b84f6acd9",
     "showTitle": false,
     "title": ""
    },
    "id": "iqo86GdmKVLL"
   },
   "outputs": [],
   "source": [
    "spark.table(\"events_per_window2\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "3ae18f87-c1df-4efa-b5ac-69c3c79555e2",
     "showTitle": false,
     "title": ""
    },
    "id": "7XWjz6W-FN6H"
   },
   "source": [
    "### Sliding Windows\n",
    "* Sliding windows can overlap if the duration of slide is smaller than the duration of window, and in this case an input can be bound to multiple windows.\n",
    "* Sliding windows use the window function with the slideDuration parameter: [pyspark.sql.functions.window](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.window.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2e508a71-ef85-4f05-8854-652ec19deefc",
     "showTitle": false,
     "title": ""
    },
    "id": "nfIKJvoUFneo"
   },
   "outputs": [],
   "source": [
    "slidingWindowQuery = sensorStreamingEvent.groupBy(window(col(\"event_time\"), windowDuration=\"10 minutes\", slideDuration=\"5 minutes\"))\\\n",
    "  .count()\\\n",
    "  .writeStream\\\n",
    "  .queryName(\"events_per_window3\")\\\n",
    "  .format(\"memory\")\\\n",
    "  .outputMode(\"complete\")\\\n",
    "  .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c22eabd5-a8cc-4dca-8adf-37d3f243fa4f",
     "showTitle": false,
     "title": ""
    },
    "id": "MzM_tYjzKdup"
   },
   "outputs": [],
   "source": [
    "spark.table(\"events_per_window3\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "4db5ddc5-20bb-4ebe-965b-52fdb42e43b4",
     "showTitle": false,
     "title": ""
    },
    "id": "NMaA2yA9_5iE"
   },
   "outputs": [],
   "source": [
    "#try with sort\n",
    "spark.table(\"events_per_window3\").sort(\"window\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "224a5c41-b90d-4245-839b-5239fc8d11be",
     "showTitle": false,
     "title": ""
    },
    "id": "89OmrqDt1jKc"
   },
   "source": [
    "### Session Windows\n",
    "* Session windows use the session_window function: [pyspark.sql.functions.session_window](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.session_window.html)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "0f1a94de-13ea-4fac-b899-c2825b73e9f6",
     "showTitle": false,
     "title": ""
    },
    "id": "CgibjPbE64d4"
   },
   "outputs": [],
   "source": [
    "sessionWindowQuery = sensorStreamingEvent.groupBy(session_window(col(\"event_time\"), gapDuration=\"10 minutes\")).count()\\\n",
    "  .writeStream\\\n",
    "  .queryName(\"events_per_window4\")\\\n",
    "  .format(\"memory\")\\\n",
    "  .outputMode(\"complete\")\\\n",
    "  .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b0145f25-ac36-4afe-b302-2ec5405ed3a7",
     "showTitle": false,
     "title": ""
    },
    "id": "u290OUCm8SW5"
   },
   "outputs": [],
   "source": [
    "spark.table(\"events_per_window4\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5ef842c1-2f78-4644-9658-2877cb49e8aa",
     "showTitle": false,
     "title": ""
    },
    "id": "hDYhrcNfHSdF"
   },
   "source": [
    "### Handling Late Data and Watermarking\n",
    "* [pyspark.sql.DataFrame.withWatermark](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withWatermark.html)  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a98d1b41-fcc1-438c-a021-a45eb7613c48",
     "showTitle": false,
     "title": ""
    },
    "id": "g7MmoLAIGGGd"
   },
   "outputs": [],
   "source": [
    "#suppose you know that your sensor data will not be late by more than 30 minutes.\n",
    "windowWithWatermarkQuery = sensorStreamingEvent\\\n",
    "  .withWatermark(\"event_time\", \"30 minutes\")\\\n",
    "  .groupBy(window(col(\"event_time\"), \"10 minutes\", \"5 minutes\"))\\\n",
    "  .count()\\\n",
    "  .writeStream\\\n",
    "  .queryName(\"events_per_window5\")\\\n",
    "  .format(\"memory\")\\\n",
    "  .outputMode(\"complete\")\\\n",
    "  .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5b34d44a-5e0f-42f1-ad20-9073e2add4b9",
     "showTitle": false,
     "title": ""
    },
    "id": "6Z633cpoKmip"
   },
   "outputs": [],
   "source": [
    "spark.table(\"events_per_window5\").show(truncate=False)"
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "language": "python",
   "notebookMetadata": {
    "mostRecentlyExecutedCommandWithImplicitDF": {
     "commandId": 4351349803017049,
     "dataframes": [
      "_sqldf"
     ]
    },
    "pythonIndentUnit": 4
   },
   "notebookName": "Streaming Aggregations",
   "widgets": {}
  },
  "colab": {
   "provenance": [
    {
     "file_id": "1-3DVy-jpDMwUjlHuvXT5nkt0fWHazHhZ",
     "timestamp": 1691337251552
    }
   ]
  },
  "kernelspec": {
   "display_name": "Python 3",
   "name": "python3"
  },
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}