4a-dashboard-generation.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Generating the input data for the realtime dashboard\n",
    "\n",
    "In this exercise, we'll use Spark structured streaming to generate the input data for the realtime dashboard."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "# Install the required Python 3 dependencies\n",
    "python3 -m pip install kafka-python  # type: ignore"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "from time import sleep\n",
    "import os\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0 pyspark-shell'\n",
    "\n",
    "from IPython.display import display, clear_output\n",
    "\n",
    "import pyspark \n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql.session import SparkSession\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.functions import *\n",
    "\n",
    "\n",
    "def test_query(sdf, mode=\"append\", rows=None, wait=2, sort=None):\n",
    "    # If a query with the same name exists, stop it.\n",
    "    query_name = \"test_query\"\n",
    "    query = None\n",
    "    for q in spark.streams.active:\n",
    "        if (q.name == query_name):\n",
    "            query = q\n",
    "    if query is not None:\n",
    "        query.stop()\n",
    "\n",
    "    try:\n",
    "        tq = (\n",
    "            # Create an output stream\n",
    "            sdf.writeStream               \n",
    "            # Only write new rows to the output\n",
    "            .outputMode(mode)           \n",
    "            # Write output stream to an in-memory Spark table (a DataFrame)\n",
    "            .format(\"memory\")               \n",
    "            # The name of the output table will be the same as the name of the query\n",
    "            .queryName(query_name)\n",
    "            # Submit the query to Spark and execute it\n",
    "            .start()\n",
    "        )\n",
    "\n",
    "        tq.processAllAvailable()\n",
    "\n",
    "        sleep(wait)\n",
    "        while(tq.status.get(\"isTriggerActive\") == True):\n",
    "            print(f\"DataAvailable: {tq.status['isDataAvailable']},\\tTriggerActive: {tq.status['isTriggerActive']}\\t{tq.status['message']}\")\n",
    "            sleep(wait)\n",
    "\n",
    "        # When the status says \"Waiting for data to arrive\", that means the query\n",
    "        # has finished its current iteration and is waiting for new messages from\n",
    "        # Kafka.\n",
    "        print(f\"DataAvailable: {tq.status['isDataAvailable']},\\tTriggerActive: {tq.status['isTriggerActive']}\\t{tq.status['message']}\")\n",
    "\n",
    "        memory_sink = spark.table(query_name)\n",
    "\n",
    "        if sort:\n",
    "            memory_sink = memory_sink.sort(*sort)\n",
    "\n",
    "        # Show result table in Jupyter Notebook. Since Jupyter Notebooks have native support for showing pandas tables,\n",
    "        # we convert the Spark DataFrame.\n",
    "        if rows:\n",
    "            display(memory_sink)\n",
    "            display(memory_sink.take(10))\n",
    "        else:\n",
    "            display(memory_sink)\n",
    "            display(memory_sink.toPandas())\n",
    "\n",
    "    finally:\n",
    "        # Always try to stop the query but it doesn't matter if it fails.\n",
    "        try:\n",
    "            tq.stop()\n",
    "        except:\n",
    "            pass\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a Spark context and specify that the python spark-kafka libraries need to be added."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a local Spark cluster with two executors (if it doesn't already exist)\n",
    "spark = SparkSession.builder.master('local[2]').getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a streaming DataFrame that represents the events received from the Kafka topic `clicks-cleaned`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input = (\n",
    "    \n",
    ")\n",
    "\n",
    "test_query(input, mode=\"append\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cast the json to columns in the DataFrame. Make sure to use TimestampType for the `ts_ingest` since we already converted it in the `clean` notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_data = (\n",
    "    \n",
    ")\n",
    "\n",
    "test_query(df_data, mode=\"append\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Generate the values you want to show in your dashboard. You are free to choose which values and aggregations to show. As an example, you can group by article title and use a 10 seconds window in order to show how many views each article received."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_data_grouped = (\n",
    "       \n",
    ")\n",
    "\n",
    "test_query(df_data_grouped, mode=\"append\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, write the results to Kafka topics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    # Remove old checkpoint dir\n",
    "    os.rmdir(\"checkpoints-dashboard\")\n",
    "except FileNotFoundError as e:\n",
    "    pass\n",
    "except OSError as e:\n",
    "    # Ignore \"file not found\" errors\n",
    "    if (e.errno != 39):\n",
    "        raise e\n",
    "\n",
    "# Prepare df for Kafka and write to kafka\n",
    "tq = (\n",
    "    \n",
    ")\n",
    "\n",
    "# Sleep two seconds\n",
    "sleep(2)\n",
    "\n",
    "# Show the status of the query\n",
    "display(tq.status)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark helpers\n",
    "\n",
    "The following code stops all running queries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for q in spark.streams.active:\n",
    "    print(\"Stopping query '{}' with name '{}'\".format(q.id, q.name))\n",
    "    q.stop()\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}