spark-streaming-openmrs.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark Stractured Streaming\n",
    "![alt text](pics/demo2.png )\n",
    "\n",
    "The motivation of this project is to provide ability of processing data in **real-time**\n",
    " from various sources like openmrs, eid, e.t.c\n",
    "\n",
    "https://spark.apache.org/docs/2.3.3/structured-streaming-kafka-integration.html#deploying\n",
    "\n",
    "https://mtpatter.github.io/bilao/notebooks/html/01-spark-struct-stream-kafka.html\n",
    "\n",
    "http://www.adaltas.com/en/2019/04/18/spark-streaming-data-pipelines-with-structured-streaming/\n",
    "\n",
    "## Set up Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.0.109:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.3.3</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Spark Structured Streaming from Kafka</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f4830690e80>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.streaming.kafka import KafkaUtils\n",
    "spark = SparkSession.builder \\\n",
    "            .appName(\"Spark Structured Streaming from Kafka\") \\\n",
    "            .config(\"spark.jars.packages\",\"org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.3\") \\\n",
    "            .config('spark.executor.memory', '10G')\\\n",
    "            .config('spark.driver.memory', '10G')\\\n",
    "            .config('spark.driver.maxResultSize', '10G')\\\n",
    "            .getOrCreate()\n",
    " \n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connection to Kakfa\n",
    "A Kafka topic can be viewed as an infinite stream where data is retained for a configurable amount of time. The infinite nature of this stream means that when starting a new query, we have to first decide what data to read and where in time we are going to begin. At a high level, there are three choices:\n",
    "\n",
    "- earliest — start reading at the beginning of the stream. This excludes data that has already been deleted from Kafka because it was older than the retention period (“aged out” data).\n",
    "- latest — start now, processing only new data that arrives after the query has started.\n",
    "\n",
    "<img src=\"https://databricks.com/wp-content/uploads/2017/04/kafka-topic.png\" width=\"300\">"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *\n",
    "import pyspark.sql.functions as f\n",
    "obs_schema = StructType([\n",
    "    StructField('obs_id', LongType(), True),\n",
    "    StructField('voided', BooleanType(), True),\n",
    "    StructField('concept_id', IntegerType(), True),\n",
    "    StructField('obs_datetime', TimestampType(), True),\n",
    "    StructField('value', StringType(), True),\n",
    "    StructField('value_type', StringType(), True),\n",
    "    StructField('obs_group_id', IntegerType(), True),\n",
    "    StructField('parent_concept_id', IntegerType(), True)\n",
    "])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Patient Stream"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- parsed_value: struct (nullable = true)\n",
      " |    |-- schema: string (nullable = true)\n",
      " |    |-- payload: struct (nullable = true)\n",
      " |    |    |-- before: string (nullable = true)\n",
      " |    |    |-- after: struct (nullable = true)\n",
      " |    |    |    |-- patient_id: long (nullable = true)\n",
      " |    |    |    |-- date_created: long (nullable = true)\n",
      " |    |    |    |-- allergy_status: string (nullable = true)\n",
      " |    |    |    |-- creator: long (nullable = true)\n",
      " |-- patient_timestamp: timestamp (nullable = true)\n",
      "\n",
      "None\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<pyspark.sql.streaming.StreamingQuery at 0x7f48305030f0>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "patient= StructType([\n",
    "    StructField('patient_id', LongType(), True),\n",
    "    StructField('date_created', LongType(), True),\n",
    "    StructField('allergy_status', StringType(), True),\n",
    "    StructField('creator', LongType(), True)\n",
    "])\n",
    "\n",
    "patientSchema= StructType([\n",
    "                StructField('schema', StringType()),\n",
    "                StructField('payload', \n",
    "                           StructType([\n",
    "                                StructField('before', StringType()),\n",
    "                                StructField('after', patient)\n",
    "                           ])\n",
    "                           )\n",
    "            ])\n",
    "\n",
    "jsonOptions = { \"timestampFormat\": \"yyyy-MM-dd'T'HH:mm:ss.sss'Z'\" }\n",
    "\n",
    "patientDF = spark \\\n",
    "    .readStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", \"localhost:9092\") \\\n",
    "    .option(\"subscribe\", \"dbserver1.openmrs.patient\") \\\n",
    "    .option(\"startingOffsets\", \"earliest\") \\\n",
    "    .load()\\\n",
    "    .select(f.from_json(f.col(\"value\").cast(\"string\"), patientSchema, jsonOptions).alias(\"parsed_value\"),\\\n",
    "            f.col(\"timestamp\").alias(\"patient_timestamp\"))\\\n",
    "    .withWatermark(\"patient_timestamp\", \"10 seconds \")   # max 10 seconds late\n",
    "    \n",
    "patientDF.createOrReplaceTempView(\"patients\")\n",
    "\n",
    "print(patientDF.printSchema())\n",
    "\n",
    "# display on console\n",
    "patientDF.select(\"parsed_value.payload.after.*\", \"patient_timestamp\").writeStream\\\n",
    "    .outputMode(\"append\")\\\n",
    "    .format(\"console\")\\\n",
    "    .option(\"truncate\", \"false\")\\\n",
    "    .start()\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![alt text](pics/patient.png )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Person Stream"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- parsed_value: struct (nullable = true)\n",
      " |    |-- schema: string (nullable = true)\n",
      " |    |-- payload: struct (nullable = true)\n",
      " |    |    |-- before: string (nullable = true)\n",
      " |    |    |-- after: struct (nullable = true)\n",
      " |    |    |    |-- person_id: long (nullable = true)\n",
      " |    |    |    |-- gender: string (nullable = true)\n",
      " |    |    |    |-- birthdate: long (nullable = true)\n",
      " |-- person_timestamp: timestamp (nullable = true)\n",
      "\n",
      "None\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<pyspark.sql.streaming.StreamingQuery at 0x7f4830516b00>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "person= StructType([\n",
    "    StructField('person_id', LongType(), True),\n",
    "    StructField('gender', StringType(), True),\n",
    "    StructField('birthdate', LongType(), True)\n",
    "])\n",
    "\n",
    "personSchema= StructType([\n",
    "                StructField('schema', StringType()),\n",
    "                StructField('payload', \n",
    "                           StructType([\n",
    "                                StructField('before', StringType()),\n",
    "                                StructField('after', person)\n",
    "                           ])\n",
    "                           )\n",
    "            ])\n",
    "\n",
    "jsonOptions = { \"timestampFormat\": \"yyyy-MM-dd'T'HH:mm:ss.sss'Z'\" }\n",
    "\n",
    "personDF = spark \\\n",
    "    .readStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", \"localhost:9092\") \\\n",
    "    .option(\"subscribe\", \"dbserver1.openmrs.person\") \\\n",
    "    .option(\"startingOffsets\", \"earliest\") \\\n",
    "    .load()\\\n",
    "    .select(f.from_json(f.col(\"value\").cast(\"string\"), personSchema, jsonOptions).alias(\"parsed_value\"),\\\n",
    "            f.col(\"timestamp\").alias(\"person_timestamp\"))\\\n",
    "    .withWatermark(\"person_timestamp\", \"10 seconds \")   # max 10 seconds late\n",
    "    \n",
    "personDF.createOrReplaceTempView(\"person\")\n",
    "\n",
    "print(personDF.printSchema())\n",
    "\n",
    "# display on console\n",
    "personDF.select(\"parsed_value.payload.after.*\", \"person_timestamp\").writeStream\\\n",
    "    .outputMode(\"append\")\\\n",
    "    .format(\"console\")\\\n",
    "    .option(\"truncate\", \"false\")\\\n",
    "    .start()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![alt text](pics/person.png )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Join the 2 streams with watermark\n",
    "![alt text](pics/join-streams.png )\n",
    "Watermarking a stream decides how delayed a record can arrive and gives a timeline after which the records can be dropped. For example, if you set a watermark for 30 minutes, then records older than 30 minutes will be dropped/ignored. When you inner join two streaming datasets watermarking and time constraint is optional. If watermark and time constraints are not specified then data is stored in the state indefinitely. Setting watermark on both sides and time constraint will enable state cleanup accordingly. Ref: https://dzone.com/articles/spark-structured-streaming-joins"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- person_id: long (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- birthdate: long (nullable = true)\n",
      " |-- person_timestamp: timestamp (nullable = true)\n",
      " |-- patient_id: long (nullable = true)\n",
      " |-- date_created: long (nullable = true)\n",
      " |-- allergy_status: string (nullable = true)\n",
      " |-- creator: long (nullable = true)\n",
      " |-- patient_timestamp: timestamp (nullable = true)\n",
      "\n",
      "None\n"
     ]
    }
   ],
   "source": [
    "personPatientDF = personDF.select(\"parsed_value.payload.after.*\", \"person_timestamp\").join(\n",
    "  patientDF.select(\"parsed_value.payload.after.*\", \"patient_timestamp\"),\n",
    "  f.expr(\"\"\" \n",
    "   person_id = patient_id AND \n",
    "    person_timestamp >= patient_timestamp AND \n",
    "    person_timestamp <= patient_timestamp + interval 10 minutes    \n",
    "    \"\"\"\n",
    "  )\n",
    ")\n",
    "\n",
    "personPatientDF.createOrReplaceTempView(\"personPatient\")\n",
    "\n",
    "print(personPatientDF.printSchema())\n",
    "\n",
    "result = personPatientDF.writeStream\\\n",
    "    .outputMode(\"append\")\\\n",
    "    .format(\"console\")\\\n",
    "    .option(\"truncate\", \"false\")\\\n",
    "    .start()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![alt text](pics/join.png )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Perform Some Aggregations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql=\"\"\"\n",
    "        SELECT\n",
    "          -- WINDOW(FROM_UNIXTIME(date_created/1000), \"1 hour\", \"1 hour\") AS eventWindow,\n",
    "          WINDOW(patient_timestamp, \"1 hour\", \"1 hour\") AS eventWindow,\n",
    "          creator,\n",
    "          Count(patient_id) AS num_events,\n",
    "          approx_count_distinct(patient_id) AS num_dist_patients,\n",
    "          AVG(patient_id) AS avgAge,\n",
    "          MIN(patient_id) AS minAge,\n",
    "          MAX(patient_id) AS maxAge\n",
    "        FROM\n",
    "          personPatient\n",
    "        GROUP BY\n",
    "          eventWindow,\n",
    "          creator\n",
    "  \"\"\"\n",
    "    \n",
    "query = spark.sql(sql)\n",
    "\n",
    "# show results\n",
    "result = query\\\n",
    "    .writeStream\\\n",
    "    .outputMode(\"append\")\\\n",
    "    .format(\"console\")\\\n",
    "    .option(\"truncate\", \"false\")\\\n",
    "    .start()\n",
    "\n",
    "\n",
    "#result.awaitTermination()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![alt text](pics/console.png )"
   ]
  }
 ],
 "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}