TopUpCustomerPredictionPart2_Step2SparkStreaming.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "67bfcf03",
   "metadata": {},
   "source": [
    "# Building Models to Predict Prospective Customers for MonPG Part 2 - Spark Streaming\n",
    "\n",
    "\n",
    "Date: 12/10/2022\n",
    "\n",
    "Version: 1.0\n",
    "\n",
    "Environment: Python 3.10.5 and Anaconda 4.11.0 (64-bit)\n",
    "\n",
    "#### Libraries used:\n",
    "\n",
    "* Spark: A fast engine for structured streaming. Combined with Python as Pyspark\n",
    "* Json to use a module called dumps for sending the streamed data to the consumer.\n",
    "* Kafka3 to produce a new topic for state-wise predictions."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7a4fbe1f",
   "metadata": {},
   "source": [
    "## 2. Streaming Application using Spark Structured Streaming\n",
    "\n",
    "In this notebook, we are going to ingest the incoming data from the two different topics `customer` and `bureau`. After ingesting it, we must bring it to an acceptable format before running it through the given ML pipeline model.\n",
    "\n",
    "**NOTE**: Please ensure that the producer code `TopUpCustomerPredictionPart2_Step1Producer.ipynb` is running before executing any writestream query here.\n",
    "\n",
    "### 2.1 Creating the Spark Session\n",
    "\n",
    "Let us start by creating a Spark session with the kafka-spark integration packages. We can specify the time zone for the Spark configuration as `UTC`. Finally, since we are dealing with a large amount of streaming data, it would be a nice idea to set the `maxtoStringFields` parameter just to make sure that a large string representation of the execution plan is published properly.\n",
    "\n",
    "Additionally, we can also set the memory Heap size to ensure that large streamed outputs are displayed without interruption.\n",
    "**Note2**: The setting of the max heap size has been done as a precautionary measure on systems with a good memory. A long-running streaming query may also fail on the Monash VM server."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "6ed9172d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importing the OS library and spark structured streaming packages\n",
    "import os\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.0.0,org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0 pyspark-shell'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "b65f8b4d",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: An illegal reflective access operation has occurred\n",
      "WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/usr/local/lib/python3.8/dist-packages/pyspark/jars/spark-unsafe_2.12-3.0.0.jar) to constructor java.nio.DirectByteBuffer(long,int)\n",
      "WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform\n",
      "WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations\n",
      "WARNING: All illegal access operations will be denied in a future release\n",
      "Ivy Default Cache set to: /srv/home/sgup0021/.ivy2/cache\n",
      "The jars for the packages stored in: /srv/home/sgup0021/.ivy2/jars\n",
      ":: loading settings :: url = jar:file:/usr/local/lib/python3.8/dist-packages/pyspark/jars/ivy-2.4.0.jar!/org/apache/ivy/core/settings/ivysettings.xml\n",
      "org.apache.spark#spark-streaming-kafka-0-10_2.12 added as a dependency\n",
      "org.apache.spark#spark-sql-kafka-0-10_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-56bb660f-ebac-4fd9-a0f7-e68f602045dc;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.apache.spark#spark-streaming-kafka-0-10_2.12;3.0.0 in central\n",
      "\tfound org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.0.0 in central\n",
      "\tfound org.apache.kafka#kafka-clients;2.4.1 in central\n",
      "\tfound com.github.luben#zstd-jni;1.4.4-3 in central\n",
      "\tfound org.lz4#lz4-java;1.7.1 in central\n",
      "\tfound org.xerial.snappy#snappy-java;1.1.7.5 in central\n",
      "\tfound org.slf4j#slf4j-api;1.7.30 in central\n",
      "\tfound org.spark-project.spark#unused;1.0.0 in central\n",
      "\tfound org.apache.spark#spark-sql-kafka-0-10_2.12;3.0.0 in central\n",
      "\tfound org.apache.commons#commons-pool2;2.6.2 in central\n",
      ":: resolution report :: resolve 474ms :: artifacts dl 9ms\n",
      "\t:: modules in use:\n",
      "\tcom.github.luben#zstd-jni;1.4.4-3 from central in [default]\n",
      "\torg.apache.commons#commons-pool2;2.6.2 from central in [default]\n",
      "\torg.apache.kafka#kafka-clients;2.4.1 from central in [default]\n",
      "\torg.apache.spark#spark-sql-kafka-0-10_2.12;3.0.0 from central in [default]\n",
      "\torg.apache.spark#spark-streaming-kafka-0-10_2.12;3.0.0 from central in [default]\n",
      "\torg.apache.spark#spark-token-provider-kafka-0-10_2.12;3.0.0 from central in [default]\n",
      "\torg.lz4#lz4-java;1.7.1 from central in [default]\n",
      "\torg.slf4j#slf4j-api;1.7.30 from central in [default]\n",
      "\torg.spark-project.spark#unused;1.0.0 from central in [default]\n",
      "\torg.xerial.snappy#snappy-java;1.1.7.5 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   10  |   0   |   0   |   0   ||   10  |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-56bb660f-ebac-4fd9-a0f7-e68f602045dc\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 10 already retrieved (0kB/14ms)\n",
      "22/10/18 19:49:40 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "# Import SparkConf class into program\n",
    "from kafka3 import KafkaProducer\n",
    "from json import dumps\n",
    "from pyspark import SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "# local[*]: run Spark in local mode with as many working processors as logical cores on your machine\n",
    "# If we want Spark to run locally with 'k' worker threads, we can specify as \"local[k]\".\n",
    "master = \"local[2]\"\n",
    "\n",
    "# The `appName` field is a name to be shown on the Spark cluster UI page\n",
    "app_name = \"Top-up Prediction Streaming Service\"\n",
    "\n",
    "# Setup configuration parameters for Spark\n",
    "spark_conf = SparkConf().setMaster(master).setAppName(app_name).set(\"spark.sql.session.timeZone\", \"UTC\")\n",
    "\n",
    "spark = SparkSession.builder.config(conf=spark_conf).config('spark.sql.debug.maxToStringFields', 5000).getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "ffe765f5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Beautifying the Jupyter notebook code by hiding warnings.\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "1718ddcb",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Hiding Spark warnings\n",
    "# Note: If faced with issues in Kafka-Spark streaming, please set log level to WARN or INFO for execution details\n",
    "spark.sparkContext.setLogLevel(\"OFF\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8b62938c",
   "metadata": {},
   "source": [
    "### 2.2 Ingesting the Streaming Data\n",
    "\n",
    "The incoming data for both streams will be ingested with the help of the `spark.readStream` function. Each stream is identified by the topic name. Additional parameters include the kafka server from which the data should be ingested. Finally setting the `startingOffsets` parameter to earliest ensures that a new query picks up from where the last one left off."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "6490bc5a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: binary (nullable = true)\n",
      " |-- value: binary (nullable = true)\n",
      " |-- topic: string (nullable = true)\n",
      " |-- partition: integer (nullable = true)\n",
      " |-- offset: long (nullable = true)\n",
      " |-- timestamp: timestamp (nullable = true)\n",
      " |-- timestampType: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Ingesting the data from the customer topic\n",
    "topic_customer = \"customer_data\"\n",
    "customer_ingest = spark \\\n",
    "    .readStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", \"127.0.0.1:9092\") \\\n",
    "    .option(\"subscribe\", topic_customer) \\\n",
    "    .option(\"startingOffsets\", 'earliest')\\\n",
    "    .load()\n",
    "\n",
    "# Printing the schema of the ingested customer topic data\n",
    "customer_ingest.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "3eed029c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: binary (nullable = true)\n",
      " |-- value: binary (nullable = true)\n",
      " |-- topic: string (nullable = true)\n",
      " |-- partition: integer (nullable = true)\n",
      " |-- offset: long (nullable = true)\n",
      " |-- timestamp: timestamp (nullable = true)\n",
      " |-- timestampType: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Ingesting the data from the bureau topic\n",
    "topic_bureau = \"bureau_data\"\n",
    "bureau_ingest = spark \\\n",
    "    .readStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", \"127.0.0.1:9092\") \\\n",
    "    .option(\"subscribe\", topic_bureau) \\\n",
    "    .option(\"startingOffsets\", 'earliest')\\\n",
    "    .load()\n",
    "\n",
    "# Printing the schema of the ingested bureau topic data\n",
    "bureau_ingest.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "78563a03",
   "metadata": {},
   "source": [
    "### 2.3 Transforming the Streamed Data\n",
    "\n",
    "With the data ingestion setup above, we can now begin the process of transforming the streamed value into something that resembles a Spark dataframe.\n",
    "\n",
    "#### 2.3.1 Defining the Schemas\n",
    "\n",
    " First, we must now prepare the schema structures for the two streams. We can specify the schema for bureau and customer by using the `ArrayType(StructType())` function to map it properly to the ingested data. We need to create two different structs for the customer data and bureau data respectively."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "7f1c9c8e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Defining the schema for the bureau data\n",
    "schema_bureau = ArrayType(StructType([\n",
    "\n",
    "    StructField('ID', StringType(), True),\n",
    "    StructField('SELF-INDICATOR', StringType(), True),\n",
    "    StructField('MATCH-TYPE', StringType(), True),\n",
    "    StructField('ACCT-TYPE', StringType(), True),\n",
    "    StructField('CONTRIBUTOR-TYPE', StringType(), True),\n",
    "    StructField('DATE-REPORTED', StringType(), True),\n",
    "    StructField('OWNERSHIP-IND', StringType(), True),\n",
    "    StructField('ACCOUNT-STATUS', StringType(), True),\n",
    "    StructField('DISBURSED-DT', StringType(), True),\n",
    "    StructField('CLOSE-DT', StringType(), True),\n",
    "    StructField('LAST-PAYMENT-DATE', StringType(), True),\n",
    "    StructField('CREDIT-LIMIT/SANC AMT', StringType(), True),\n",
    "    StructField('DISBURSED-AMT/HIGH CREDIT', StringType(), True),\n",
    "    StructField('INSTALLMENT-AMT', StringType(), True),\n",
    "    StructField('CURRENT-BAL', StringType(), True),\n",
    "    StructField('INSTALLMENT-FREQUENCY', StringType(), True),\n",
    "    StructField('OVERDUE-AMT', StringType(), True),\n",
    "    StructField('WRITE-OFF-AMT', StringType(), True),\n",
    "    StructField('ASSET_CLASS', StringType(), True),\n",
    "    StructField('REPORTED DATE - HIST', StringType(), True),\n",
    "    StructField('DPD - HIST', StringType(), True),\n",
    "    StructField('CUR BAL - HIST', StringType(), True),\n",
    "    StructField('AMT OVERDUE - HIST', StringType(), True),\n",
    "    StructField('AMT PAID - HIST', StringType(), True),\n",
    "    StructField('TENURE', StringType(), True),\n",
    "    StructField('ts', TimestampType(), True)\n",
    "]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "8df3c758",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Defining the schema for the customer data\n",
    "schema_customer = ArrayType(StructType([\n",
    "\n",
    "    StructField('ID', StringType(), True),\n",
    "    StructField('Frequency', StringType(), True),\n",
    "    StructField('InstlmentMode', StringType(), True),\n",
    "    StructField('LoanStatus', StringType(), True),\n",
    "    StructField('PaymentMode', StringType(), True),\n",
    "    StructField('BranchID', StringType(), True),\n",
    "    StructField('Area', StringType(), True),\n",
    "    StructField('Tenure', StringType(), True),\n",
    "    StructField('AssetCost', StringType(), True),\n",
    "    StructField('AmountFinance', StringType(), True),\n",
    "    StructField('DisbursalAmount', StringType(), True),\n",
    "    StructField('EMI', StringType(), True),\n",
    "    StructField('DisbursalDate', StringType(), True),\n",
    "    StructField('MaturityDAte', StringType(), True),\n",
    "    StructField('AuthDate', StringType(), True),\n",
    "    StructField('AssetID', StringType(), True),\n",
    "    StructField('ManufacturerID', StringType(), True),\n",
    "    StructField('SupplierID', StringType(), True),\n",
    "    StructField('LTV', StringType(), True),\n",
    "    StructField('SEX', StringType(), True),\n",
    "    StructField('AGE', StringType(), True),\n",
    "    StructField('MonthlyIncome', StringType(), True),\n",
    "    StructField('City', StringType(), True),\n",
    "    StructField('State', StringType(), True),\n",
    "    StructField('ZiPCODE', StringType(), True),\n",
    "    StructField('Top-up Month', StringType(), True),\n",
    "    StructField('ts', TimestampType(), True)\n",
    "]))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "14825b9a",
   "metadata": {},
   "source": [
    "#### 2.3.2 Parsing the Data\n",
    "\n",
    "With the structs defined for both streams, we will use them to accept the ingested streams and give them structure. This will be accomplished in three steps:\n",
    "\n",
    "* Ingest the value part of the batch message from json and store it in string representation.\n",
    "* Explode the structure of the parsed value to expose the unnested data.\n",
    "* Query the unnested struct and extract individual columns. We can also cast the appropriate datatype for some of these columns. For the columns that must be converted to integer datatype and contain non-numerical characters in them, we will handle them separately."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "64172318",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- parsed_value: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- ID: string (nullable = true)\n",
      " |    |    |-- Frequency: string (nullable = true)\n",
      " |    |    |-- InstlmentMode: string (nullable = true)\n",
      " |    |    |-- LoanStatus: string (nullable = true)\n",
      " |    |    |-- PaymentMode: string (nullable = true)\n",
      " |    |    |-- BranchID: string (nullable = true)\n",
      " |    |    |-- Area: string (nullable = true)\n",
      " |    |    |-- Tenure: string (nullable = true)\n",
      " |    |    |-- AssetCost: string (nullable = true)\n",
      " |    |    |-- AmountFinance: string (nullable = true)\n",
      " |    |    |-- DisbursalAmount: string (nullable = true)\n",
      " |    |    |-- EMI: string (nullable = true)\n",
      " |    |    |-- DisbursalDate: string (nullable = true)\n",
      " |    |    |-- MaturityDAte: string (nullable = true)\n",
      " |    |    |-- AuthDate: string (nullable = true)\n",
      " |    |    |-- AssetID: string (nullable = true)\n",
      " |    |    |-- ManufacturerID: string (nullable = true)\n",
      " |    |    |-- SupplierID: string (nullable = true)\n",
      " |    |    |-- LTV: string (nullable = true)\n",
      " |    |    |-- SEX: string (nullable = true)\n",
      " |    |    |-- AGE: string (nullable = true)\n",
      " |    |    |-- MonthlyIncome: string (nullable = true)\n",
      " |    |    |-- City: string (nullable = true)\n",
      " |    |    |-- State: string (nullable = true)\n",
      " |    |    |-- ZiPCODE: string (nullable = true)\n",
      " |    |    |-- Top-up Month: string (nullable = true)\n",
      " |    |    |-- ts: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Parsing the ingested customer data in json format\n",
    "customer_json = customer_ingest.select(F.from_json(F.col(\"value\").cast(\"string\"), schema_customer).alias('parsed_value'))\n",
    "\n",
    "# Checking the schema of the parsed value\n",
    "customer_json.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "3e2b2be9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- unnested_value: struct (nullable = true)\n",
      " |    |-- ID: string (nullable = true)\n",
      " |    |-- Frequency: string (nullable = true)\n",
      " |    |-- InstlmentMode: string (nullable = true)\n",
      " |    |-- LoanStatus: string (nullable = true)\n",
      " |    |-- PaymentMode: string (nullable = true)\n",
      " |    |-- BranchID: string (nullable = true)\n",
      " |    |-- Area: string (nullable = true)\n",
      " |    |-- Tenure: string (nullable = true)\n",
      " |    |-- AssetCost: string (nullable = true)\n",
      " |    |-- AmountFinance: string (nullable = true)\n",
      " |    |-- DisbursalAmount: string (nullable = true)\n",
      " |    |-- EMI: string (nullable = true)\n",
      " |    |-- DisbursalDate: string (nullable = true)\n",
      " |    |-- MaturityDAte: string (nullable = true)\n",
      " |    |-- AuthDate: string (nullable = true)\n",
      " |    |-- AssetID: string (nullable = true)\n",
      " |    |-- ManufacturerID: string (nullable = true)\n",
      " |    |-- SupplierID: string (nullable = true)\n",
      " |    |-- LTV: string (nullable = true)\n",
      " |    |-- SEX: string (nullable = true)\n",
      " |    |-- AGE: string (nullable = true)\n",
      " |    |-- MonthlyIncome: string (nullable = true)\n",
      " |    |-- City: string (nullable = true)\n",
      " |    |-- State: string (nullable = true)\n",
      " |    |-- ZiPCODE: string (nullable = true)\n",
      " |    |-- Top-up Month: string (nullable = true)\n",
      " |    |-- ts: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Unnesting the json extracted customer data \n",
    "customer_explode = customer_json.select(F.explode(F.col(\"parsed_value\")).alias('unnested_value'))\n",
    "\n",
    "# Checking the schema of the unnested customer data\n",
    "customer_explode.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "8cb5cc5e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ID: string (nullable = true)\n",
      " |-- Frequency: string (nullable = true)\n",
      " |-- InstlmentMode: string (nullable = true)\n",
      " |-- LoanStatus: string (nullable = true)\n",
      " |-- PaymentMode: string (nullable = true)\n",
      " |-- BranchID: string (nullable = true)\n",
      " |-- Area: string (nullable = true)\n",
      " |-- Tenure: integer (nullable = true)\n",
      " |-- AssetCost: integer (nullable = true)\n",
      " |-- AmountFinance: double (nullable = true)\n",
      " |-- DisbursalAmount: double (nullable = true)\n",
      " |-- EMI: double (nullable = true)\n",
      " |-- DisbursalDate: timestamp (nullable = true)\n",
      " |-- MaturityDate: timestamp (nullable = true)\n",
      " |-- AuthDate: timestamp (nullable = true)\n",
      " |-- AssetID: string (nullable = true)\n",
      " |-- ManufacturerID: string (nullable = true)\n",
      " |-- SupplierID: string (nullable = true)\n",
      " |-- LTV: double (nullable = true)\n",
      " |-- Sex: string (nullable = true)\n",
      " |-- Age: integer (nullable = true)\n",
      " |-- MonthlyIncome: double (nullable = true)\n",
      " |-- City: string (nullable = true)\n",
      " |-- State: string (nullable = true)\n",
      " |-- ZipCode: string (nullable = true)\n",
      " |-- Top-up Month: string (nullable = true)\n",
      " |-- ts: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Extracting the customer columns from the unnested struct\n",
    "customer_df_formatted = customer_explode.select(\n",
    "    F.col(\"unnested_value.ID\").alias(\"ID\"),\n",
    "    F.col(\"unnested_value.Frequency\").alias(\"Frequency\"),\n",
    "    F.col(\"unnested_value.InstlmentMode\").alias(\"InstlmentMode\"),\n",
    "    F.col(\"unnested_value.LoanStatus\").alias(\"LoanStatus\"),\n",
    "    F.col(\"unnested_value.PaymentMode\").alias(\"PaymentMode\"),\n",
    "    F.col(\"unnested_value.BranchID\").alias(\"BranchID\"),\n",
    "    F.col(\"unnested_value.Area\").alias(\"Area\"),\n",
    "    F.col(\"unnested_value.Tenure\").cast('int').alias(\"Tenure\"),\n",
    "    F.col(\"unnested_value.AssetCost\").cast('int').alias(\"AssetCost\"),\n",
    "    F.col(\"unnested_value.AmountFinance\").cast('double').alias(\"AmountFinance\"),\n",
    "    F.col(\"unnested_value.DisbursalAmount\").cast('double').alias(\"DisbursalAmount\"),\n",
    "    F.col(\"unnested_value.EMI\").cast('double').alias(\"EMI\"),\n",
    "    F.col(\"unnested_value.DisbursalDate\").cast('timestamp').alias(\"DisbursalDate\"),\n",
    "    F.col(\"unnested_value.MaturityDate\").cast('timestamp').alias(\"MaturityDate\"),\n",
    "    F.col(\"unnested_value.AuthDate\").cast('timestamp').alias(\"AuthDate\"),\n",
    "    F.col(\"unnested_value.AssetID\").alias(\"AssetID\"),\n",
    "    F.col(\"unnested_value.ManufacturerID\").alias(\"ManufacturerID\"),\n",
    "    F.col(\"unnested_value.SupplierID\").alias(\"SupplierID\"),\n",
    "    F.col(\"unnested_value.LTV\").cast('double').alias(\"LTV\"),\n",
    "    F.col(\"unnested_value.Sex\").alias(\"Sex\"),\n",
    "    F.col(\"unnested_value.Age\").cast('int').alias(\"Age\"),\n",
    "    F.col(\"unnested_value.MonthlyIncome\").cast('double').alias(\"MonthlyIncome\"),\n",
    "    F.col(\"unnested_value.City\").alias(\"City\"),\n",
    "    F.col(\"unnested_value.State\").alias(\"State\"),\n",
    "    F.col(\"unnested_value.ZipCode\").alias(\"ZipCode\"),\n",
    "    F.col(\"unnested_value.Top-up Month\").alias(\"Top-up Month\"),\n",
    "    F.col(\"unnested_value.ts\").alias(\"ts\")\n",
    ")\n",
    "\n",
    "# Checking the final formatted schema for the customer data\n",
    "customer_df_formatted.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4c6650b0",
   "metadata": {},
   "source": [
    "Now that we are done extracting the customer data, let us confirm that we have received the data in the correct form. We want to make sure that we have the proper data to be used in the subsequent aggregations. We can use `writeStream` to print the streaming data to the console. To avoid memory issues with the VM, let us select a few columns only."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "83a1e382",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 7:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 0\n",
      "-------------------------------------------\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------------------+-----------------+\n",
      "| ID|                 ts|     Top-Up Month|\n",
      "+---+-------------------+-----------------+\n",
      "|  1|2022-10-13 21:58:18|      > 48 Months|\n",
      "|  2|2022-10-13 21:58:18|No Top-up Service|\n",
      "|  3|2022-10-13 21:58:18|     12-18 Months|\n",
      "|  4|2022-10-13 21:58:18|     36-48 Months|\n",
      "|  5|2022-10-13 21:58:18|     18-24 Months|\n",
      "|  6|2022-10-13 21:58:18|     30-36 Months|\n",
      "|  7|2022-10-13 21:58:18|      > 48 Months|\n",
      "|  8|2022-10-13 21:58:18|     36-48 Months|\n",
      "|  9|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 10|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 11|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 12|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 13|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 14|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 15|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 16|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 17|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 18|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 19|2022-10-13 21:58:18|No Top-up Service|\n",
      "| 20|2022-10-13 21:58:18|No Top-up Service|\n",
      "+---+-------------------+-----------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 1\n",
      "-------------------------------------------\n",
      "+---+-------------------+-----------------+\n",
      "| ID|                 ts|     Top-Up Month|\n",
      "+---+-------------------+-----------------+\n",
      "|285|2022-10-17 01:12:40|No Top-up Service|\n",
      "|286|2022-10-17 01:12:40|No Top-up Service|\n",
      "|287|2022-10-17 01:12:40|No Top-up Service|\n",
      "|288|2022-10-17 01:12:40|      > 48 Months|\n",
      "|289|2022-10-17 01:12:40|No Top-up Service|\n",
      "|290|2022-10-17 01:12:40|No Top-up Service|\n",
      "|291|2022-10-17 01:12:40|No Top-up Service|\n",
      "|292|2022-10-17 01:12:40|No Top-up Service|\n",
      "|293|2022-10-17 01:12:40|No Top-up Service|\n",
      "|294|2022-10-17 01:12:40|No Top-up Service|\n",
      "|295|2022-10-17 01:12:40|No Top-up Service|\n",
      "|296|2022-10-17 01:12:40|No Top-up Service|\n",
      "+---+-------------------+-----------------+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 2\n",
      "-------------------------------------------\n",
      "+---+-------------------+-----------------+\n",
      "| ID|                 ts|     Top-Up Month|\n",
      "+---+-------------------+-----------------+\n",
      "|297|2022-10-17 01:12:51|No Top-up Service|\n",
      "|298|2022-10-17 01:12:51|No Top-up Service|\n",
      "|299|2022-10-17 01:12:51|No Top-up Service|\n",
      "|300|2022-10-17 01:12:51|No Top-up Service|\n",
      "|301|2022-10-17 01:12:51|      > 48 Months|\n",
      "|302|2022-10-17 01:12:51|No Top-up Service|\n",
      "|303|2022-10-17 01:12:51|No Top-up Service|\n",
      "|304|2022-10-17 01:12:51|No Top-up Service|\n",
      "|305|2022-10-17 01:12:51|No Top-up Service|\n",
      "|306|2022-10-17 01:12:51|No Top-up Service|\n",
      "|307|2022-10-17 01:12:51|No Top-up Service|\n",
      "|308|2022-10-17 01:12:51|No Top-up Service|\n",
      "|309|2022-10-17 01:12:51|No Top-up Service|\n",
      "|310|2022-10-17 01:12:51|No Top-up Service|\n",
      "|311|2022-10-17 01:12:51|No Top-up Service|\n",
      "|312|2022-10-17 01:12:51|No Top-up Service|\n",
      "|313|2022-10-17 01:12:51|No Top-up Service|\n",
      "|314|2022-10-17 01:12:51|No Top-up Service|\n",
      "|315|2022-10-17 01:12:51|No Top-up Service|\n",
      "|316|2022-10-17 01:12:51|No Top-up Service|\n",
      "+---+-------------------+-----------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Checking the formatted customer data\n",
    "query = customer_df_formatted.select('ID', 'ts', 'Top-Up Month')\\\n",
    "    .writeStream \\\n",
    "    .outputMode(\"append\") \\\n",
    "    .format(\"console\") \\\n",
    "    .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "63d52f93",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stopping the stream\n",
    "query.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "20b5352b",
   "metadata": {},
   "source": [
    "We will take a similar approach for the bureau data as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "541197f7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- parsed_value: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- ID: string (nullable = true)\n",
      " |    |    |-- SELF-INDICATOR: string (nullable = true)\n",
      " |    |    |-- MATCH-TYPE: string (nullable = true)\n",
      " |    |    |-- ACCT-TYPE: string (nullable = true)\n",
      " |    |    |-- CONTRIBUTOR-TYPE: string (nullable = true)\n",
      " |    |    |-- DATE-REPORTED: string (nullable = true)\n",
      " |    |    |-- OWNERSHIP-IND: string (nullable = true)\n",
      " |    |    |-- ACCOUNT-STATUS: string (nullable = true)\n",
      " |    |    |-- DISBURSED-DT: string (nullable = true)\n",
      " |    |    |-- CLOSE-DT: string (nullable = true)\n",
      " |    |    |-- LAST-PAYMENT-DATE: string (nullable = true)\n",
      " |    |    |-- CREDIT-LIMIT/SANC AMT: string (nullable = true)\n",
      " |    |    |-- DISBURSED-AMT/HIGH CREDIT: string (nullable = true)\n",
      " |    |    |-- INSTALLMENT-AMT: string (nullable = true)\n",
      " |    |    |-- CURRENT-BAL: string (nullable = true)\n",
      " |    |    |-- INSTALLMENT-FREQUENCY: string (nullable = true)\n",
      " |    |    |-- OVERDUE-AMT: string (nullable = true)\n",
      " |    |    |-- WRITE-OFF-AMT: string (nullable = true)\n",
      " |    |    |-- ASSET_CLASS: string (nullable = true)\n",
      " |    |    |-- REPORTED DATE - HIST: string (nullable = true)\n",
      " |    |    |-- DPD - HIST: string (nullable = true)\n",
      " |    |    |-- CUR BAL - HIST: string (nullable = true)\n",
      " |    |    |-- AMT OVERDUE - HIST: string (nullable = true)\n",
      " |    |    |-- AMT PAID - HIST: string (nullable = true)\n",
      " |    |    |-- TENURE: string (nullable = true)\n",
      " |    |    |-- ts: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Parsing the ingested bureau data in json format\n",
    "bureau_json = bureau_ingest.select(F.from_json(F.col(\"value\").cast(\"string\"), schema_bureau).alias('parsed_value'))\n",
    "\n",
    "# Checking the schema of the parsed value\n",
    "bureau_json.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "c6309301",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- unnested_value: struct (nullable = true)\n",
      " |    |-- ID: string (nullable = true)\n",
      " |    |-- SELF-INDICATOR: string (nullable = true)\n",
      " |    |-- MATCH-TYPE: string (nullable = true)\n",
      " |    |-- ACCT-TYPE: string (nullable = true)\n",
      " |    |-- CONTRIBUTOR-TYPE: string (nullable = true)\n",
      " |    |-- DATE-REPORTED: string (nullable = true)\n",
      " |    |-- OWNERSHIP-IND: string (nullable = true)\n",
      " |    |-- ACCOUNT-STATUS: string (nullable = true)\n",
      " |    |-- DISBURSED-DT: string (nullable = true)\n",
      " |    |-- CLOSE-DT: string (nullable = true)\n",
      " |    |-- LAST-PAYMENT-DATE: string (nullable = true)\n",
      " |    |-- CREDIT-LIMIT/SANC AMT: string (nullable = true)\n",
      " |    |-- DISBURSED-AMT/HIGH CREDIT: string (nullable = true)\n",
      " |    |-- INSTALLMENT-AMT: string (nullable = true)\n",
      " |    |-- CURRENT-BAL: string (nullable = true)\n",
      " |    |-- INSTALLMENT-FREQUENCY: string (nullable = true)\n",
      " |    |-- OVERDUE-AMT: string (nullable = true)\n",
      " |    |-- WRITE-OFF-AMT: string (nullable = true)\n",
      " |    |-- ASSET_CLASS: string (nullable = true)\n",
      " |    |-- REPORTED DATE - HIST: string (nullable = true)\n",
      " |    |-- DPD - HIST: string (nullable = true)\n",
      " |    |-- CUR BAL - HIST: string (nullable = true)\n",
      " |    |-- AMT OVERDUE - HIST: string (nullable = true)\n",
      " |    |-- AMT PAID - HIST: string (nullable = true)\n",
      " |    |-- TENURE: string (nullable = true)\n",
      " |    |-- ts: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Unnesting the json extracted bureau data \n",
    "bureau_explode = bureau_json.select(F.explode(F.col(\"parsed_value\")).alias('unnested_value'))\n",
    "\n",
    "# Checking the schema of the unnested bureau data\n",
    "bureau_explode.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "017213d1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ID: string (nullable = true)\n",
      " |-- SELF-INDICATOR: string (nullable = true)\n",
      " |-- MATCH-TYPE: string (nullable = true)\n",
      " |-- ACCT-TYPE: string (nullable = true)\n",
      " |-- CONTRIBUTOR-TYPE: string (nullable = true)\n",
      " |-- DATE-REPORTED: timestamp (nullable = true)\n",
      " |-- OWNERSHIP-IND: string (nullable = true)\n",
      " |-- ACCOUNT-STATUS: string (nullable = true)\n",
      " |-- DISBURSED-DT: timestamp (nullable = true)\n",
      " |-- CLOSE-DT: timestamp (nullable = true)\n",
      " |-- LAST-PAYMENT-DATE: timestamp (nullable = true)\n",
      " |-- CREDIT-LIMIT/SANC AMT: string (nullable = true)\n",
      " |-- DISBURSED-AMT/HIGH CREDIT: string (nullable = true)\n",
      " |-- INSTALLMENT-AMT: string (nullable = true)\n",
      " |-- CURRENT-BAL: string (nullable = true)\n",
      " |-- INSTALLMENT-FREQUENCY: string (nullable = true)\n",
      " |-- OVERDUE-AMT: string (nullable = true)\n",
      " |-- WRITE-OFF-AMT: integer (nullable = true)\n",
      " |-- ASSET_CLASS: string (nullable = true)\n",
      " |-- REPORTED DATE - HIST: string (nullable = true)\n",
      " |-- DPD - HIST: string (nullable = true)\n",
      " |-- CUR BAL - HIST: string (nullable = true)\n",
      " |-- AMT OVERDUE - HIST: string (nullable = true)\n",
      " |-- AMT PAID - HIST: string (nullable = true)\n",
      " |-- TENURE: string (nullable = true)\n",
      " |-- ts: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Extracting the bureau columns from the unnested struct\n",
    "bureau_df_formatted = bureau_explode.select(\n",
    "    F.col(\"unnested_value.ID\").alias(\"ID\"),\n",
    "    F.col(\"unnested_value.SELF-INDICATOR\").alias(\"SELF-INDICATOR\"),\n",
    "    F.col(\"unnested_value.MATCH-TYPE\").alias(\"MATCH-TYPE\"),\n",
    "    F.col(\"unnested_value.ACCT-TYPE\").alias(\"ACCT-TYPE\"),\n",
    "    F.col(\"unnested_value.CONTRIBUTOR-TYPE\").alias(\"CONTRIBUTOR-TYPE\"),\n",
    "    F.col(\"unnested_value.DATE-REPORTED\").cast('timestamp').alias(\"DATE-REPORTED\"),\n",
    "    F.col(\"unnested_value.OWNERSHIP-IND\").alias(\"OWNERSHIP-IND\"),\n",
    "    F.col(\"unnested_value.ACCOUNT-STATUS\").alias(\"ACCOUNT-STATUS\"),\n",
    "    F.col(\"unnested_value.DISBURSED-DT\").cast('timestamp').alias(\"DISBURSED-DT\"),\n",
    "    F.col(\"unnested_value.CLOSE-DT\").cast('timestamp').alias(\"CLOSE-DT\"),\n",
    "    F.col(\"unnested_value.LAST-PAYMENT-DATE\").cast('timestamp').alias(\"LAST-PAYMENT-DATE\"),\n",
    "    F.col(\"unnested_value.CREDIT-LIMIT/SANC AMT\").alias(\"CREDIT-LIMIT/SANC AMT\"),\n",
    "    F.col(\"unnested_value.DISBURSED-AMT/HIGH CREDIT\").alias(\"DISBURSED-AMT/HIGH CREDIT\"),\n",
    "    F.col(\"unnested_value.INSTALLMENT-AMT\").alias(\"INSTALLMENT-AMT\"),\n",
    "    F.col(\"unnested_value.CURRENT-BAL\").alias(\"CURRENT-BAL\"),\n",
    "    F.col(\"unnested_value.INSTALLMENT-FREQUENCY\").alias(\"INSTALLMENT-FREQUENCY\"),\n",
    "    F.col(\"unnested_value.OVERDUE-AMT\").alias(\"OVERDUE-AMT\"),\n",
    "    F.col(\"unnested_value.WRITE-OFF-AMT\").cast('int').alias(\"WRITE-OFF-AMT\"),\n",
    "    F.col(\"unnested_value.ASSET_CLASS\").alias(\"ASSET_CLASS\"),\n",
    "    F.col(\"unnested_value.REPORTED DATE - HIST\").alias(\"REPORTED DATE - HIST\"),\n",
    "    F.col(\"unnested_value.DPD - HIST\").alias(\"DPD - HIST\"),\n",
    "    F.col(\"unnested_value.CUR BAL - HIST\").alias(\"CUR BAL - HIST\"),\n",
    "    F.col(\"unnested_value.AMT OVERDUE - HIST\").alias(\"AMT OVERDUE - HIST\"),\n",
    "    F.col(\"unnested_value.AMT PAID - HIST\").alias(\"AMT PAID - HIST\"),\n",
    "    F.col(\"unnested_value.TENURE\").alias(\"TENURE\"),\n",
    "    F.col(\"unnested_value.ts\").alias(\"ts\")\n",
    ")\n",
    "\n",
    "# Checking the schema of the formatted bureau data\n",
    "bureau_df_formatted.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9c14d7f2",
   "metadata": {},
   "source": [
    "In the final step of setting up the bureau data, we need to change the datatype of some of its columns to integer. However, since they also contain non-numeric characters, we must remove them before converting the datatype of these columns. The columns to be converted are:\n",
    "\n",
    "* `DISBURSED-AMT/HIGH CREDIT`\n",
    "* `CURRENT-BAL`\n",
    "* `OVERDUE-AMT`\n",
    "* `INSTALLMENT-AMT`\n",
    "\n",
    "The code block to be used for this section was implemented effectively in 2A, and we will use it here too."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "0abc9ea4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Defining a udf function to discard non-numeric characters\n",
    "udf_num = F.udf(lambda a: ''.join(b for b in a if b.isdigit()) if a else '')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "80f772ff",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Applying the function to the disbursed amount column\n",
    "bureau_df_formatted = bureau_df_formatted.withColumn('DISBURSED-AMT/HIGH CREDIT', udf_num('DISBURSED-AMT/HIGH CREDIT'))\n",
    "\n",
    "# Converting the column data type to integer\n",
    "bureau_df_formatted = bureau_df_formatted.withColumn(\"DISBURSED-AMT/HIGH CREDIT\", \n",
    "                                 bureau_df_formatted[\"DISBURSED-AMT/HIGH CREDIT\"].cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "0d6397da",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Applying the transformation function to the current balance column\n",
    "bureau_df_formatted = bureau_df_formatted.withColumn('CURRENT-BAL', udf_num('CURRENT-BAL'))\n",
    "\n",
    "# Converting the column data type to integer\n",
    "bureau_df_formatted = bureau_df_formatted.withColumn(\"CURRENT-BAL\", \n",
    "                                 bureau_df_formatted[\"CURRENT-BAL\"].cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "e8a3a8ef",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Applying the transformation function to the overdue amount column\n",
    "bureau_df_formatted = bureau_df_formatted.withColumn('OVERDUE-AMT', udf_num('OVERDUE-AMT'))\n",
    "\n",
    "# Converting the column data type to integer \n",
    "bureau_df_formatted = bureau_df_formatted.withColumn(\"OVERDUE-AMT\", \n",
    "                                 bureau_df_formatted[\"OVERDUE-AMT\"].cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "68ca5c97",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Applying the transformation function to the installment amount column\n",
    "bureau_df_formatted = bureau_df_formatted.withColumn(\"INSTALLMENT-AMT\", udf_num('INSTALLMENT-AMT'))\n",
    "\n",
    "# Updating the datatype for the installment amount column\n",
    "bureau_df_formatted = bureau_df_formatted.withColumn(\"INSTALLMENT-AMT\", \n",
    "                                                     bureau_df_formatted[\"INSTALLMENT-AMT\"].cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "7f0cee48",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ID: string (nullable = true)\n",
      " |-- SELF-INDICATOR: string (nullable = true)\n",
      " |-- MATCH-TYPE: string (nullable = true)\n",
      " |-- ACCT-TYPE: string (nullable = true)\n",
      " |-- CONTRIBUTOR-TYPE: string (nullable = true)\n",
      " |-- DATE-REPORTED: timestamp (nullable = true)\n",
      " |-- OWNERSHIP-IND: string (nullable = true)\n",
      " |-- ACCOUNT-STATUS: string (nullable = true)\n",
      " |-- DISBURSED-DT: timestamp (nullable = true)\n",
      " |-- CLOSE-DT: timestamp (nullable = true)\n",
      " |-- LAST-PAYMENT-DATE: timestamp (nullable = true)\n",
      " |-- CREDIT-LIMIT/SANC AMT: string (nullable = true)\n",
      " |-- DISBURSED-AMT/HIGH CREDIT: integer (nullable = true)\n",
      " |-- INSTALLMENT-AMT: integer (nullable = true)\n",
      " |-- CURRENT-BAL: integer (nullable = true)\n",
      " |-- INSTALLMENT-FREQUENCY: string (nullable = true)\n",
      " |-- OVERDUE-AMT: integer (nullable = true)\n",
      " |-- WRITE-OFF-AMT: integer (nullable = true)\n",
      " |-- ASSET_CLASS: string (nullable = true)\n",
      " |-- REPORTED DATE - HIST: string (nullable = true)\n",
      " |-- DPD - HIST: string (nullable = true)\n",
      " |-- CUR BAL - HIST: string (nullable = true)\n",
      " |-- AMT OVERDUE - HIST: string (nullable = true)\n",
      " |-- AMT PAID - HIST: string (nullable = true)\n",
      " |-- TENURE: string (nullable = true)\n",
      " |-- ts: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Checking the schema of the formatted bureau data\n",
    "bureau_df_formatted.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7f43d98a",
   "metadata": {},
   "source": [
    "From the schema, it looks like the formatting process was a success. Now, let us use `writestream` to print the stream data to the console to confirm that there are no issues. Once again, let us limit the number of columns to some of the important ones that we just converted in the previous step."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "e7597a93",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 0\n",
      "-------------------------------------------\n",
      "+---+-------------------+-----------+-------------------------+\n",
      "| ID|                 ts|OVERDUE-AMT|DISBURSED-AMT/HIGH CREDIT|\n",
      "+---+-------------------+-----------+-------------------------+\n",
      "|  1|2022-10-13 21:58:18|       null|                    44000|\n",
      "|  1|2022-10-13 21:58:18|      37873|                    37352|\n",
      "|  1|2022-10-13 21:58:18|          0|                   400000|\n",
      "|  1|2022-10-13 21:58:18|          0|                   500000|\n",
      "|  1|2022-10-13 21:58:18|          0|                   145000|\n",
      "|  1|2022-10-13 21:58:18|       null|                        0|\n",
      "|  1|2022-10-13 21:58:18|          0|                   300000|\n",
      "|  1|2022-10-13 21:58:18|       null|                   500000|\n",
      "|  1|2022-10-13 21:58:18|          0|                   275000|\n",
      "|  2|2022-10-13 21:58:18|          0|                   450000|\n",
      "|  2|2022-10-13 21:58:18|          0|                   354176|\n",
      "|  2|2022-10-13 21:58:18|       null|                   300000|\n",
      "|  2|2022-10-13 21:58:18|          0|                  1700000|\n",
      "|  2|2022-10-13 21:58:18|          0|                   400000|\n",
      "|  2|2022-10-13 21:58:18|          0|                  1100000|\n",
      "|  2|2022-10-13 21:58:18|          0|                   350000|\n",
      "|  2|2022-10-13 21:58:18|          0|                  3000000|\n",
      "|  2|2022-10-13 21:58:18|          0|                   785905|\n",
      "|  2|2022-10-13 21:58:18|          0|                   700000|\n",
      "|  2|2022-10-13 21:58:18|          0|                  3800000|\n",
      "+---+-------------------+-----------+-------------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 1\n",
      "-------------------------------------------\n",
      "+---+-------------------+-----------+-------------------------+\n",
      "| ID|                 ts|OVERDUE-AMT|DISBURSED-AMT/HIGH CREDIT|\n",
      "+---+-------------------+-----------+-------------------------+\n",
      "|979|2022-10-17 01:19:04|          0|                   300000|\n",
      "|979|2022-10-17 01:19:04|          0|                    40000|\n",
      "|979|2022-10-17 01:19:04|       1555|                    40000|\n",
      "|980|2022-10-17 01:19:04|          0|                   350000|\n",
      "|981|2022-10-17 01:19:04|          0|                   917742|\n",
      "|981|2022-10-17 01:19:04|       null|                   150000|\n",
      "|981|2022-10-17 01:19:04|          0|                   325000|\n",
      "|981|2022-10-17 01:19:04|       null|                   180000|\n",
      "|981|2022-10-17 01:19:04|       null|                    34900|\n",
      "|982|2022-10-17 01:19:04|          0|                  1279178|\n",
      "|982|2022-10-17 01:19:04|          0|                  1205316|\n",
      "|982|2022-10-17 01:19:04|          0|                    42992|\n",
      "|982|2022-10-17 01:19:04|          0|                   900000|\n",
      "|982|2022-10-17 01:19:04|          0|                   300000|\n",
      "|982|2022-10-17 01:19:04|          0|                   900000|\n",
      "|982|2022-10-17 01:19:04|          0|                    68843|\n",
      "|982|2022-10-17 01:19:04|          0|                   300000|\n",
      "|982|2022-10-17 01:19:04|          0|                     2574|\n",
      "|982|2022-10-17 01:19:04|          0|                   739500|\n",
      "|982|2022-10-17 01:19:04|          0|                   350000|\n",
      "+---+-------------------+-----------+-------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Checking the formatted bureau data\n",
    "query = bureau_df_formatted.select('ID', 'ts', 'OVERDUE-AMT', 'DISBURSED-AMT/HIGH CREDIT')\\\n",
    "    .writeStream \\\n",
    "    .outputMode(\"append\") \\\n",
    "    .format(\"console\") \\\n",
    "    .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "a83a5544",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stopping the query\n",
    "query.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3e67de2c",
   "metadata": {},
   "source": [
    "#### 2.3.3 Watermarking\n",
    "\n",
    "In the final step of our data preparation, we need to apply a watermark to both streams. Watermarking is a useful feature in Spark that handles the data that arrives late. Spark handles the late data by storing it in memory and including it in any future aggregations or manipulations that are done to the stream. The watermark can be set with the `withWatermark()` function. It requires a timestamp value which the column `ts` will provide. We will set the delay threshold to 5 seconds."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "64d4c807",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Watermarking the customer stream\n",
    "watermarked_customer = customer_df_formatted.withWatermark(\"ts\", \"5 seconds\")\n",
    "\n",
    "# Watermarking the bureau stream\n",
    "watermarked_bureau = bureau_df_formatted.withWatermark(\"ts\", \"5 seconds\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f39bce64",
   "metadata": {},
   "source": [
    "### 2.4 Grouping the Bureau Data\n",
    "\n",
    "The below actions will make some of the same changes that were made to the bureau data in 2A. The major operation includes grouping the numeric and non-numeric columns of the bureau data respectively.\n",
    "\n",
    "#### 2.4.1 Transforming `SELF-INDICATOR`\n",
    "\n",
    "In the first step, we will transform the values of the column `SELF-INDICATOR` from True/False to 1/0 respectively."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "980525ce",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Temporarily renaming the column for convenience of running the next query\n",
    "watermarked_bureau = watermarked_bureau.withColumnRenamed('SELF-INDICATOR', 'self')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "deb31ed6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Transforming the values of SELF-INDICATOR\n",
    "watermarked_bureau = watermarked_bureau.withColumn(\"self\", F.when(watermarked_bureau.self == 'TRUE', 1) \\\n",
    "      .when(watermarked_bureau.self == 'FALSE', 0) \\\n",
    "      .otherwise(watermarked_bureau.self))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "61f88378",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reverting to the original name of the column\n",
    "watermarked_bureau = watermarked_bureau.withColumnRenamed('self', 'SELF-INDICATOR')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "b4d731f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Changing the datatype of the column to integer\n",
    "watermarked_bureau = watermarked_bureau.withColumn(\"SELF-INDICATOR\", F.col(\"SELF-INDICATOR\").cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "fdb7ac18",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ID: string (nullable = true)\n",
      " |-- SELF-INDICATOR: integer (nullable = true)\n",
      " |-- MATCH-TYPE: string (nullable = true)\n",
      " |-- ACCT-TYPE: string (nullable = true)\n",
      " |-- CONTRIBUTOR-TYPE: string (nullable = true)\n",
      " |-- DATE-REPORTED: timestamp (nullable = true)\n",
      " |-- OWNERSHIP-IND: string (nullable = true)\n",
      " |-- ACCOUNT-STATUS: string (nullable = true)\n",
      " |-- DISBURSED-DT: timestamp (nullable = true)\n",
      " |-- CLOSE-DT: timestamp (nullable = true)\n",
      " |-- LAST-PAYMENT-DATE: timestamp (nullable = true)\n",
      " |-- CREDIT-LIMIT/SANC AMT: string (nullable = true)\n",
      " |-- DISBURSED-AMT/HIGH CREDIT: integer (nullable = true)\n",
      " |-- INSTALLMENT-AMT: integer (nullable = true)\n",
      " |-- CURRENT-BAL: integer (nullable = true)\n",
      " |-- INSTALLMENT-FREQUENCY: string (nullable = true)\n",
      " |-- OVERDUE-AMT: integer (nullable = true)\n",
      " |-- WRITE-OFF-AMT: integer (nullable = true)\n",
      " |-- ASSET_CLASS: string (nullable = true)\n",
      " |-- REPORTED DATE - HIST: string (nullable = true)\n",
      " |-- DPD - HIST: string (nullable = true)\n",
      " |-- CUR BAL - HIST: string (nullable = true)\n",
      " |-- AMT OVERDUE - HIST: string (nullable = true)\n",
      " |-- AMT PAID - HIST: string (nullable = true)\n",
      " |-- TENURE: string (nullable = true)\n",
      " |-- ts: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Checking the schema of the transformed column\n",
    "watermarked_bureau.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ffebf597",
   "metadata": {},
   "source": [
    "#### 2.4.2 Grouping the Bureau Data\n",
    "\n",
    "We will use the same grouping operation from 2A to aggregate the numeric and non-numeric columns of the bureau data. This means that, for the numeric columns, we will aggregate them by `summing` the values and give the postfix `_sum`. The non-numeric columns will be aggregated by computing their distinct `counts`. However, we have two changes in the code here:\n",
    "\n",
    "* Firstly, we need to aggregate the columns simultaneously, given that Spark streaming does not support multiple aggregations yet.\n",
    "* For the non-numeric columns, we will apply `approx_count_distinct` to it instead of `countDistinct` as Spark streaming does not support the former.\n",
    "* Finally, we will also provide a 30-second window duration for the aggregation. Specifying a window will put the aggregated rows into time bound intervals."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "20366236",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating a new dataframe with aggregated sum of numeric columns for each ID and aggregated count of non-numeric columns for each ID.\n",
    "bureau_groups = watermarked_bureau.groupBy(F.window(watermarked_bureau.ts, \"30 seconds\"), F.col(\"ID\")).agg(*(F.sum(F.col(column.name)).alias(column.name + '_sum')\n",
    "                                                        for column in watermarked_bureau.schema.fields if column.name != 'ID' and\n",
    "                                                        (isinstance(column.dataType, IntegerType) or\n",
    "                                                         isinstance(column.dataType, DoubleType))), *(F.approx_count_distinct(F.col(column.name)).alias(column.name + '_dist')\n",
    "                                                            for column in watermarked_bureau.schema.fields if column.name != 'ID' and column.name != 'ts' and\n",
    "                                                            (not isinstance(column.dataType, IntegerType) or\n",
    "                                                             isinstance(column.dataType, DoubleType))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "23dd8b8b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- window: struct (nullable = false)\n",
      " |    |-- start: timestamp (nullable = true)\n",
      " |    |-- end: timestamp (nullable = true)\n",
      " |-- ID: string (nullable = true)\n",
      " |-- SELF-INDICATOR_sum: long (nullable = true)\n",
      " |-- DISBURSED-AMT/HIGH CREDIT_sum: long (nullable = true)\n",
      " |-- INSTALLMENT-AMT_sum: long (nullable = true)\n",
      " |-- CURRENT-BAL_sum: long (nullable = true)\n",
      " |-- OVERDUE-AMT_sum: long (nullable = true)\n",
      " |-- WRITE-OFF-AMT_sum: long (nullable = true)\n",
      " |-- MATCH-TYPE_dist: long (nullable = false)\n",
      " |-- ACCT-TYPE_dist: long (nullable = false)\n",
      " |-- CONTRIBUTOR-TYPE_dist: long (nullable = false)\n",
      " |-- DATE-REPORTED_dist: long (nullable = false)\n",
      " |-- OWNERSHIP-IND_dist: long (nullable = false)\n",
      " |-- ACCOUNT-STATUS_dist: long (nullable = false)\n",
      " |-- DISBURSED-DT_dist: long (nullable = false)\n",
      " |-- CLOSE-DT_dist: long (nullable = false)\n",
      " |-- LAST-PAYMENT-DATE_dist: long (nullable = false)\n",
      " |-- CREDIT-LIMIT/SANC AMT_dist: long (nullable = false)\n",
      " |-- INSTALLMENT-FREQUENCY_dist: long (nullable = false)\n",
      " |-- ASSET_CLASS_dist: long (nullable = false)\n",
      " |-- REPORTED DATE - HIST_dist: long (nullable = false)\n",
      " |-- DPD - HIST_dist: long (nullable = false)\n",
      " |-- CUR BAL - HIST_dist: long (nullable = false)\n",
      " |-- AMT OVERDUE - HIST_dist: long (nullable = false)\n",
      " |-- AMT PAID - HIST_dist: long (nullable = false)\n",
      " |-- TENURE_dist: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Checking the schema of the bureau grouped data\n",
    "bureau_groups.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f9c31031",
   "metadata": {},
   "source": [
    "### 2.5 Unnesting the Window Columns and Joining the Streams\n",
    "\n",
    "We are going to perform two operations in this step.\n",
    "\n",
    "* Firstly, we need to unnest the window struct into two columns called `window_start` and `window_end` respectively.\n",
    "* Finally, we will join the grouped bureau data with the customer data based on the ID column.\n",
    "\n",
    "#### 2.5.1 Unnesting the Window Columns\n",
    "\n",
    "We can unnest the window struct into the window start and end columns simply by using the `select()`, `col()` and 'alias()` functions as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "dd8c7fe0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Extracting the window start and end columns from the window struct\n",
    "bureau_groups = bureau_groups.select(*bureau_groups.columns[1:], \n",
    "                                                   F.col(\"window.start\").alias(\"window_start\"), \n",
    "                                                   F.col(\"window.end\").alias(\"window_end\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "1c30174b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ID: string (nullable = true)\n",
      " |-- SELF-INDICATOR_sum: long (nullable = true)\n",
      " |-- DISBURSED-AMT/HIGH CREDIT_sum: long (nullable = true)\n",
      " |-- INSTALLMENT-AMT_sum: long (nullable = true)\n",
      " |-- CURRENT-BAL_sum: long (nullable = true)\n",
      " |-- OVERDUE-AMT_sum: long (nullable = true)\n",
      " |-- WRITE-OFF-AMT_sum: long (nullable = true)\n",
      " |-- MATCH-TYPE_dist: long (nullable = false)\n",
      " |-- ACCT-TYPE_dist: long (nullable = false)\n",
      " |-- CONTRIBUTOR-TYPE_dist: long (nullable = false)\n",
      " |-- DATE-REPORTED_dist: long (nullable = false)\n",
      " |-- OWNERSHIP-IND_dist: long (nullable = false)\n",
      " |-- ACCOUNT-STATUS_dist: long (nullable = false)\n",
      " |-- DISBURSED-DT_dist: long (nullable = false)\n",
      " |-- CLOSE-DT_dist: long (nullable = false)\n",
      " |-- LAST-PAYMENT-DATE_dist: long (nullable = false)\n",
      " |-- CREDIT-LIMIT/SANC AMT_dist: long (nullable = false)\n",
      " |-- INSTALLMENT-FREQUENCY_dist: long (nullable = false)\n",
      " |-- ASSET_CLASS_dist: long (nullable = false)\n",
      " |-- REPORTED DATE - HIST_dist: long (nullable = false)\n",
      " |-- DPD - HIST_dist: long (nullable = false)\n",
      " |-- CUR BAL - HIST_dist: long (nullable = false)\n",
      " |-- AMT OVERDUE - HIST_dist: long (nullable = false)\n",
      " |-- AMT PAID - HIST_dist: long (nullable = false)\n",
      " |-- TENURE_dist: long (nullable = false)\n",
      " |-- window_start: timestamp (nullable = true)\n",
      " |-- window_end: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Printing the final schema for bureau groups\n",
    "bureau_groups.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2e25f81e",
   "metadata": {},
   "source": [
    "Let us use writestream to check a few columns of the grouped data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "id": "938f10c7",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 0\n",
      "-------------------------------------------\n",
      "+---+-----------------------------+------------------+------------+----------+\n",
      "| ID|DISBURSED-AMT/HIGH CREDIT_sum|DATE-REPORTED_dist|window_start|window_end|\n",
      "+---+-----------------------------+------------------+------------+----------+\n",
      "+---+-----------------------------+------------------+------------+----------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 1\n",
      "-------------------------------------------\n",
      "+----+-----------------------------+------------------+-------------------+-------------------+\n",
      "|  ID|DISBURSED-AMT/HIGH CREDIT_sum|DATE-REPORTED_dist|       window_start|         window_end|\n",
      "+----+-----------------------------+------------------+-------------------+-------------------+\n",
      "| 811|                      1340000|                 5|2022-10-14 06:03:00|2022-10-14 06:03:30|\n",
      "|2811|                      1129000|                 3|2022-10-15 03:00:00|2022-10-15 03:00:30|\n",
      "|  57|                     70734744|                33|2022-10-14 06:28:00|2022-10-14 06:28:30|\n",
      "|2660|                      1130000|                 3|2022-10-15 04:39:00|2022-10-15 04:39:30|\n",
      "|5519|                      3552611|                11|2022-10-14 04:19:30|2022-10-14 04:20:00|\n",
      "|1579|                       250000|                 1|2022-10-17 01:40:30|2022-10-17 01:41:00|\n",
      "| 389|                      4090000|                 5|2022-10-15 08:50:30|2022-10-15 08:51:00|\n",
      "|4128|                       360000|                 1|2022-10-14 07:33:30|2022-10-14 07:34:00|\n",
      "|5082|                       200000|                 1|2022-10-14 07:40:30|2022-10-14 07:41:00|\n",
      "| 242|                      8757392|                 9|2022-10-15 03:55:00|2022-10-15 03:55:30|\n",
      "|3372|                       515000|                 1|2022-10-15 03:04:00|2022-10-15 03:04:30|\n",
      "|4070|                       514269|                 3|2022-10-15 05:04:30|2022-10-15 05:05:00|\n",
      "|2707|                       400000|                 1|2022-10-14 03:52:30|2022-10-14 03:53:00|\n",
      "| 199|                       873000|                 6|2022-10-17 01:11:30|2022-10-17 01:12:00|\n",
      "|3787|                      1120400|                 2|2022-10-14 04:03:00|2022-10-14 04:03:30|\n",
      "|7172|                       429000|                 2|2022-10-14 04:35:30|2022-10-14 04:36:00|\n",
      "|1063|                     48577656|                10|2022-10-15 12:22:00|2022-10-15 12:22:30|\n",
      "|5243|                       350000|                 1|2022-10-14 04:17:00|2022-10-14 04:17:30|\n",
      "|2604|                       300000|                 1|2022-10-17 01:50:00|2022-10-17 01:50:30|\n",
      "|  49|                      1707866|                 8|2022-10-14 03:27:00|2022-10-14 03:27:30|\n",
      "+----+-----------------------------+------------------+-------------------+-------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 8:================>                                       (60 + 2) / 200]\r"
     ]
    }
   ],
   "source": [
    "# Checking the formatted bureau grouped data\n",
    "query = bureau_groups.select('ID', 'DISBURSED-AMT/HIGH CREDIT_sum', 'DATE-REPORTED_dist', \n",
    "                             'window_start', 'window_end')\\\n",
    "    .writeStream \\\n",
    "    .outputMode(\"append\") \\\n",
    "    .format(\"console\") \\\n",
    "    .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "id": "a7c80967",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stopping the stream\n",
    "query.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2764d67c",
   "metadata": {},
   "source": [
    "### 2.5.2 Joining the Streams\n",
    "\n",
    "We will join the grouped bureau stream with the customer stream by using the `join()` function in Spark and `expr()` to specify the joining condition. Additionally, we also need to include only those joined rows where the timestamp of the customer row lies between the window start time and window end time of the corresponding bureau data. To specify these two conditions, we can use the `alias()` function for convenience.\n",
    "\n",
    "First, let us create a copy of the customer stream and rename the ID column so that we may be able to drop the extra ID column easily from the joined data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "197bb536",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Renaming the ID column in the customer dataset for conveniently dropping it from the joined data\n",
    "customer_new = watermarked_customer.withColumnRenamed(\"ID\",\"ID2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "a728ed16",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Joining the customer and bureau streams\n",
    "cust_bureau_df = bureau_groups.alias('bu').join(customer_new.alias('cust'), on = F.expr(\"\"\"\n",
    "cust.ID2 = bu.ID AND\n",
    "cust.ts BETWEEN bu.window_start AND bu.window_end\n",
    "\"\"\"), how = \"inner\").drop('ID2')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "35138df7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ID: string (nullable = true)\n",
      " |-- SELF-INDICATOR_sum: long (nullable = true)\n",
      " |-- DISBURSED-AMT/HIGH CREDIT_sum: long (nullable = true)\n",
      " |-- INSTALLMENT-AMT_sum: long (nullable = true)\n",
      " |-- CURRENT-BAL_sum: long (nullable = true)\n",
      " |-- OVERDUE-AMT_sum: long (nullable = true)\n",
      " |-- WRITE-OFF-AMT_sum: long (nullable = true)\n",
      " |-- MATCH-TYPE_dist: long (nullable = false)\n",
      " |-- ACCT-TYPE_dist: long (nullable = false)\n",
      " |-- CONTRIBUTOR-TYPE_dist: long (nullable = false)\n",
      " |-- DATE-REPORTED_dist: long (nullable = false)\n",
      " |-- OWNERSHIP-IND_dist: long (nullable = false)\n",
      " |-- ACCOUNT-STATUS_dist: long (nullable = false)\n",
      " |-- DISBURSED-DT_dist: long (nullable = false)\n",
      " |-- CLOSE-DT_dist: long (nullable = false)\n",
      " |-- LAST-PAYMENT-DATE_dist: long (nullable = false)\n",
      " |-- CREDIT-LIMIT/SANC AMT_dist: long (nullable = false)\n",
      " |-- INSTALLMENT-FREQUENCY_dist: long (nullable = false)\n",
      " |-- ASSET_CLASS_dist: long (nullable = false)\n",
      " |-- REPORTED DATE - HIST_dist: long (nullable = false)\n",
      " |-- DPD - HIST_dist: long (nullable = false)\n",
      " |-- CUR BAL - HIST_dist: long (nullable = false)\n",
      " |-- AMT OVERDUE - HIST_dist: long (nullable = false)\n",
      " |-- AMT PAID - HIST_dist: long (nullable = false)\n",
      " |-- TENURE_dist: long (nullable = false)\n",
      " |-- window_start: timestamp (nullable = true)\n",
      " |-- window_end: timestamp (nullable = true)\n",
      " |-- Frequency: string (nullable = true)\n",
      " |-- InstlmentMode: string (nullable = true)\n",
      " |-- LoanStatus: string (nullable = true)\n",
      " |-- PaymentMode: string (nullable = true)\n",
      " |-- BranchID: string (nullable = true)\n",
      " |-- Area: string (nullable = true)\n",
      " |-- Tenure: integer (nullable = true)\n",
      " |-- AssetCost: integer (nullable = true)\n",
      " |-- AmountFinance: double (nullable = true)\n",
      " |-- DisbursalAmount: double (nullable = true)\n",
      " |-- EMI: double (nullable = true)\n",
      " |-- DisbursalDate: timestamp (nullable = true)\n",
      " |-- MaturityDate: timestamp (nullable = true)\n",
      " |-- AuthDate: timestamp (nullable = true)\n",
      " |-- AssetID: string (nullable = true)\n",
      " |-- ManufacturerID: string (nullable = true)\n",
      " |-- SupplierID: string (nullable = true)\n",
      " |-- LTV: double (nullable = true)\n",
      " |-- Sex: string (nullable = true)\n",
      " |-- Age: integer (nullable = true)\n",
      " |-- MonthlyIncome: double (nullable = true)\n",
      " |-- City: string (nullable = true)\n",
      " |-- State: string (nullable = true)\n",
      " |-- ZipCode: string (nullable = true)\n",
      " |-- Top-up Month: string (nullable = true)\n",
      " |-- ts: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Checking the schema of the final joined stream\n",
    "cust_bureau_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f0cb8597",
   "metadata": {},
   "source": [
    "### 2.6 Writing to Parquet\n",
    "\n",
    "Parquet is a file format that is supported by Spark. Parquet files are able to maintain the information about the schema of the data automatically. However, unlike csv files or txt files, parquet files are not readable until they are loaded with Spark's special capabilities.\n",
    "\n",
    "For the above created customer bureau stream, we can use `writestream` to persist the stream data in parquet format. We will also specify the checkpoint location for the parquet files. As the name suggests, it will contain the state of the streamed data for fault tolerance. Let us write some data to parquet format and stop the writestream after we have persisted sufficient data. The columns to be selected for persisting to parquet are:\n",
    "\n",
    "* `ID`\n",
    "* `window_start`\n",
    "* `window_end`\n",
    "* `ts`\n",
    "* `Top-up Month` which will be renamed to `Top-up_Month`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "id": "c92b2a84",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 50:=======================>                               (87 + 2) / 200]\r"
     ]
    }
   ],
   "source": [
    "# Writing the data into parquet format\n",
    "query_file_sink = cust_bureau_df.select('ID', 'window_start', 'window_end', 'ts', 'Top-up Month')\\\n",
    "                                .withColumnRenamed('Top-up Month', 'Top-up_Month')\\\n",
    "                                .writeStream.format(\"parquet\")\\\n",
    "                                .outputMode(\"append\")\\\n",
    "                                .option(\"path\", \"parquet/cust_bureau_df\")\\\n",
    "                                .option(\"checkpointLocation\", \"parquet/cust_bureau_df/checkpoint\")\\\n",
    "                                .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "id": "19ea34f6",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 50:==========================>                            (97 + 2) / 200]\r"
     ]
    }
   ],
   "source": [
    "# Stopping the file_sink query\n",
    "query_file_sink.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "47905fba",
   "metadata": {},
   "source": [
    "### 2.7 Pipeline Model Transformation and Parquet Persistence\n",
    "\n",
    "This section is a continuation of the activities performed in 2A. In 2A, we were able to successfully persist a machine learning model pipeline. In this section, we are going to check whether a pipeline model can be applied to the customer bureau stream. We need to perform two actions here:\n",
    "\n",
    "* Transform the data with the loaded model pipeline.\n",
    "* Persist the model predictions in parquet format.\n",
    "\n",
    "#### 2.7.1 Making Predictions\n",
    "\n",
    "Firstly, let us import the `PipelineModel` module of the pyspark.ml library."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "3ec25453",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importing the required libraries\n",
    "from pyspark.ml import PipelineModel"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8603d28a",
   "metadata": {},
   "source": [
    "Next, we can load the persisted pipeline model. The function `load()` will allow us to do this."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "5e4c2ad5",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Loading the pipeline model\n",
    "gradient_booster_fit = PipelineModel.load(\"topup_pipeline_model\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc511174",
   "metadata": {},
   "source": [
    "Unlike the data we dealt with in 2A, this new customer bureau data may contain null values. Therefore, we must st the model configuration for the NULL handling to `keep` here. We will set this configuration for stage 2 of the model which is the Vector Assembler."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "1e29e6cb",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "VectorAssembler_ac48f700b95d"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Adjusting the loaded model to handle NULL values\n",
    "gradient_booster_fit.stages[-2].setHandleInvalid(\"keep\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6eadfe0d",
   "metadata": {},
   "source": [
    "We can now use the familiar `transform()` function to apply the model to the customer bureau stream and generate the predictions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "360f72f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Applying the fitted model to the customer bureau stream\n",
    "gb_pred = gradient_booster_fit.transform(cust_bureau_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "796608a5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ID: string (nullable = true)\n",
      " |-- SELF-INDICATOR_sum: long (nullable = true)\n",
      " |-- DISBURSED-AMT/HIGH CREDIT_sum: long (nullable = true)\n",
      " |-- INSTALLMENT-AMT_sum: long (nullable = true)\n",
      " |-- CURRENT-BAL_sum: long (nullable = true)\n",
      " |-- OVERDUE-AMT_sum: long (nullable = true)\n",
      " |-- WRITE-OFF-AMT_sum: long (nullable = true)\n",
      " |-- MATCH-TYPE_dist: long (nullable = false)\n",
      " |-- ACCT-TYPE_dist: long (nullable = false)\n",
      " |-- CONTRIBUTOR-TYPE_dist: long (nullable = false)\n",
      " |-- DATE-REPORTED_dist: long (nullable = false)\n",
      " |-- OWNERSHIP-IND_dist: long (nullable = false)\n",
      " |-- ACCOUNT-STATUS_dist: long (nullable = false)\n",
      " |-- DISBURSED-DT_dist: long (nullable = false)\n",
      " |-- CLOSE-DT_dist: long (nullable = false)\n",
      " |-- LAST-PAYMENT-DATE_dist: long (nullable = false)\n",
      " |-- CREDIT-LIMIT/SANC AMT_dist: long (nullable = false)\n",
      " |-- INSTALLMENT-FREQUENCY_dist: long (nullable = false)\n",
      " |-- ASSET_CLASS_dist: long (nullable = false)\n",
      " |-- REPORTED DATE - HIST_dist: long (nullable = false)\n",
      " |-- DPD - HIST_dist: long (nullable = false)\n",
      " |-- CUR BAL - HIST_dist: long (nullable = false)\n",
      " |-- AMT OVERDUE - HIST_dist: long (nullable = false)\n",
      " |-- AMT PAID - HIST_dist: long (nullable = false)\n",
      " |-- TENURE_dist: long (nullable = false)\n",
      " |-- window_start: timestamp (nullable = true)\n",
      " |-- window_end: timestamp (nullable = true)\n",
      " |-- Frequency: string (nullable = true)\n",
      " |-- InstlmentMode: string (nullable = true)\n",
      " |-- LoanStatus: string (nullable = true)\n",
      " |-- PaymentMode: string (nullable = true)\n",
      " |-- BranchID: string (nullable = true)\n",
      " |-- Area: string (nullable = true)\n",
      " |-- Tenure: integer (nullable = true)\n",
      " |-- AssetCost: integer (nullable = true)\n",
      " |-- AmountFinance: double (nullable = true)\n",
      " |-- DisbursalAmount: double (nullable = true)\n",
      " |-- EMI: double (nullable = true)\n",
      " |-- DisbursalDate: timestamp (nullable = true)\n",
      " |-- MaturityDate: timestamp (nullable = true)\n",
      " |-- AuthDate: timestamp (nullable = true)\n",
      " |-- AssetID: string (nullable = true)\n",
      " |-- ManufacturerID: string (nullable = true)\n",
      " |-- SupplierID: string (nullable = true)\n",
      " |-- LTV: double (nullable = true)\n",
      " |-- Sex: string (nullable = true)\n",
      " |-- Age: integer (nullable = true)\n",
      " |-- MonthlyIncome: double (nullable = true)\n",
      " |-- City: string (nullable = true)\n",
      " |-- State: string (nullable = true)\n",
      " |-- ZipCode: string (nullable = true)\n",
      " |-- Top-up Month: string (nullable = true)\n",
      " |-- ts: timestamp (nullable = true)\n",
      " |-- State_idx: double (nullable = false)\n",
      " |-- City_idx: double (nullable = false)\n",
      " |-- BranchID_idx: double (nullable = false)\n",
      " |-- Area_idx: double (nullable = false)\n",
      " |-- ManufacturerID_idx: double (nullable = false)\n",
      " |-- LoanStatus_idx: double (nullable = false)\n",
      " |-- Tenure_idx: double (nullable = false)\n",
      " |-- City_vec: vector (nullable = true)\n",
      " |-- State_vec: vector (nullable = true)\n",
      " |-- Tenure_vec: vector (nullable = true)\n",
      " |-- LoanStatus_vec: vector (nullable = true)\n",
      " |-- Area_vec: vector (nullable = true)\n",
      " |-- BranchID_vec: vector (nullable = true)\n",
      " |-- ManufacturerID_vec: vector (nullable = true)\n",
      " |-- features: vector (nullable = true)\n",
      " |-- rawPrediction: vector (nullable = true)\n",
      " |-- probability: vector (nullable = true)\n",
      " |-- prediction: double (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Printing the schema of the predicted stream\n",
    "gb_pred.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5ba33420",
   "metadata": {},
   "source": [
    "#### 2.7.2 Persisting the Predictions\n",
    "\n",
    "If we observe the schema of the transformed data, we can see that the predictions have been successfully generated here. However, let us confirm that the data in these predictions is proper with the help of the writestream function of Spark streaming. We will specifically select those columns to be displayed on the console which we intend to write to the parquet file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "id": "fb938aba",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 0\n",
      "-------------------------------------------\n",
      "+---+------------+----------+---+----------+------------+\n",
      "| ID|window_start|window_end| ts|prediction|Top-up_Month|\n",
      "+---+------------+----------+---+----------+------------+\n",
      "+---+------------+----------+---+----------+------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 1\n",
      "-------------------------------------------\n",
      "+----+-------------------+-------------------+-------------------+----------+-----------------+\n",
      "|  ID|       window_start|         window_end|                 ts|prediction|     Top-up_Month|\n",
      "+----+-------------------+-------------------+-------------------+----------+-----------------+\n",
      "|3414|2022-10-15 04:52:30|2022-10-15 04:53:00|2022-10-15 04:52:53|       1.0|No Top-up Service|\n",
      "|1159|2022-10-14 06:06:00|2022-10-14 06:06:30|2022-10-14 06:06:10|       0.0|No Top-up Service|\n",
      "|1436|2022-10-15 02:50:30|2022-10-15 02:51:00|2022-10-15 02:50:43|       0.0|No Top-up Service|\n",
      "| 675|2022-10-15 09:33:30|2022-10-15 09:34:00|2022-10-15 09:33:41|       0.0|No Top-up Service|\n",
      "| 675|2022-10-15 08:53:30|2022-10-15 08:54:00|2022-10-15 08:53:47|       0.0|No Top-up Service|\n",
      "|1159|2022-10-14 05:37:30|2022-10-14 05:38:00|2022-10-14 05:37:38|       0.0|No Top-up Service|\n",
      "|4032|2022-10-14 04:05:30|2022-10-14 04:06:00|2022-10-14 04:05:40|       1.0|      > 48 Months|\n",
      "|4032|2022-10-15 05:04:00|2022-10-15 05:04:30|2022-10-15 05:04:09|       1.0|      > 48 Months|\n",
      "|5645|2022-10-14 04:21:00|2022-10-14 04:21:30|2022-10-14 04:21:10|       0.0|No Top-up Service|\n",
      "| 829|2022-10-14 06:03:00|2022-10-14 06:03:30|2022-10-14 06:03:14|       0.0|No Top-up Service|\n",
      "|2904|2022-10-14 03:54:30|2022-10-14 03:55:00|2022-10-14 03:54:43|       0.0|No Top-up Service|\n",
      "|1436|2022-10-14 05:40:30|2022-10-14 05:41:00|2022-10-14 05:40:32|       0.0|No Top-up Service|\n",
      "|1159|2022-10-15 12:23:30|2022-10-15 12:24:00|2022-10-15 12:23:49|       0.0|No Top-up Service|\n",
      "|1090|2022-10-14 07:12:30|2022-10-14 07:13:00|2022-10-14 07:12:37|       0.0|     24-30 Months|\n",
      "| 691|2022-10-13 22:03:30|2022-10-13 22:04:00|2022-10-13 22:03:59|       1.0|     24-30 Months|\n",
      "| 467|2022-10-15 03:59:00|2022-10-15 03:59:30|2022-10-15 03:59:24|       0.0|No Top-up Service|\n",
      "|1159|2022-10-15 08:59:00|2022-10-15 08:59:30|2022-10-15 08:59:17|       0.0|No Top-up Service|\n",
      "| 296|2022-10-15 12:10:30|2022-10-15 12:11:00|2022-10-15 12:10:30|       0.0|No Top-up Service|\n",
      "| 691|2022-10-14 06:43:30|2022-10-14 06:44:00|2022-10-14 06:43:48|       1.0|     24-30 Months|\n",
      "|1159|2022-10-15 02:48:30|2022-10-15 02:49:00|2022-10-15 02:48:43|       0.0|No Top-up Service|\n",
      "+----+-------------------+-------------------+-------------------+----------+-----------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 81:=======================>                               (87 + 2) / 200]\r"
     ]
    }
   ],
   "source": [
    "# Checking the prediction data\n",
    "query = gb_pred.withColumnRenamed('Top-up Month', \n",
    "                                    'Top-up_Month').select('ID', \n",
    "                                                           'window_start', \n",
    "                                                           'window_end', \n",
    "                                                           'ts', \n",
    "                                                           'prediction', 'Top-up_Month')\\\n",
    "                                                   .writeStream \\\n",
    "                                                   .outputMode(\"append\") \\\n",
    "                                                   .format(\"console\") \\\n",
    "                                                   .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "id": "49d91525",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 81:==========================>                            (96 + 2) / 200]\r"
     ]
    }
   ],
   "source": [
    "query.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "50035707",
   "metadata": {},
   "source": [
    "It looks like the predictions are coming along nicely. We can go ahead and persist these predictions in parquet format. Similar to what we did in section 2.6, we will rename the column `Top-up Month` as `Top-up_Month`. As always, we will let a decent number of parquet files be written before we stop the query."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "id": "cc028b5c",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 91:===================================================>  (189 + 2) / 200]\r"
     ]
    }
   ],
   "source": [
    "# Persisting the prediction data in parquet format\n",
    "query_file_sink = gb_pred.withColumnRenamed('Top-up Month', 'Top-up_Month')\\\n",
    "                         .select('ID', 'window_start', 'window_end', 'ts', 'prediction', 'Top-up_Month')\\\n",
    "                         .writeStream.format(\"parquet\")\\\n",
    "                         .outputMode(\"append\")\\\n",
    "                         .option(\"path\", \"./parquet/top_up_predictions\")\\\n",
    "                         .option(\"checkpointLocation\", \"./parquet/top_up_predictions/checkpoint\")\\\n",
    "                         .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "id": "b5ea3101",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 91:====================================================> (195 + 2) / 200]\r"
     ]
    }
   ],
   "source": [
    "# Stop the file_sink query\n",
    "query_file_sink.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a0d18006",
   "metadata": {},
   "source": [
    "### 2.8 Target Customers\n",
    "\n",
    "We have confirmed that we are able to successfully generate predictions for the data. In this section, we will be focusing on the target customers from these predictions. In other words, these are the predictions where the Top-up prediction is 1. We want to know more about these target customers with respect to the state they come from. Furthermore, we need to broadcast this information as a kafka topic to the consumer.\n",
    "\n",
    "Therefore, we will group the predictions by State and count them. Since we have a number of moving parts here, let us define a function to be applied to each batch of the customer bureau stream and send it over to the consumer. Here are the steps that will be followed:\n",
    "\n",
    "* Check if the current customer bureau batch is empty. If it is empty then skip to the next batch.\n",
    "* If the batch is not empty, transform it with the pipeline model and generate the predictions.\n",
    "* Initialise a new Kafka producer on the same bootstrap sever with a `value_serializer` to convert the data into a stream of bytes and a `key_serializer` to do the same to the key object.\n",
    "* Filter the predictions by target customers, group them by state and count them.\n",
    "* Check if the grouping of the current batch is empty or not. If it is empty, then it will skip to the next batch.\n",
    "* Find the max timestamp of the `window_end` time.\n",
    "* Convert the aggregated prediction data to json as a preparation for streaming.\n",
    "* Publish the topic and its data to the consumer.\n",
    "* Print the published data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "f905e56e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Defining a function to send the predicted customer data to the consumer\n",
    "def prediction_stream(cust_bureau, epochID):\n",
    "    \n",
    "    # Checking if the current batch contains any rows\n",
    "    if cust_bureau.count() == 0:\n",
    "\n",
    "        print(\"Current batch is empty! Skipping..\")\n",
    "\n",
    "        # Skipping\n",
    "        return\n",
    "    \n",
    "    # Opening a try block\n",
    "    try:\n",
    "        \n",
    "        # Making predictions with the pipeline model\n",
    "        predictionStream_df = gradient_booster_fit.transform(cust_bureau)\n",
    "\n",
    "        prediction_topic = \"Statewide_customer_count\"\n",
    "        \n",
    "        # Initialising a kafka producer to send the prediction data to the consumer\n",
    "        prediction_producer = KafkaProducer(bootstrap_servers = ['localhost:9092'], \n",
    "                                  value_serializer = lambda v: dumps(v).encode('ascii'),\n",
    "                                  key_serializer = lambda a: dumps(a).encode('ascii'),\n",
    "                                   api_version = (0, 10))\n",
    "        \n",
    "        # Isolating the predictions in terms of the target consumers\n",
    "        state_count_df = predictionStream_df.filter(\"prediction = 1.0\")\\\n",
    "                                            .groupBy(\"State\").count()\\\n",
    "                                            .withColumnRenamed(\"count\",\"target_customers\")\n",
    "\n",
    "        # Checking if the aggregated prediction data is empty\n",
    "        if state_count_df.count() == 0:\n",
    "\n",
    "            print(\"No update to customer counts in the current batch! Skipping..\")\n",
    "\n",
    "            # Skipping\n",
    "            return\n",
    "\n",
    "        else:\n",
    "\n",
    "            # Applying the window_end value as the key for the current batch\n",
    "            end_ts = str(predictionStream_df.agg({\"window_end\":\"max\"}).collect()[0][0])\n",
    "        \n",
    "            # Setting up the data to be streamed\n",
    "            stream_data = state_count_df.toPandas().to_json(orient = \"records\")\n",
    "        \n",
    "            # Publishing the streamed data\n",
    "            prediction_producer.send(prediction_topic, value = stream_data, key = end_ts)\n",
    "        \n",
    "            # Printing the data being streamed\n",
    "            print(f\"Epoch ID: {epochID}, Timestamp: {end_ts},  Data: {stream_data}\")\n",
    "        \n",
    "    except Exception as ex:\n",
    "        \n",
    "        print(f\"Exception observed in current batch {ex}. Exiting...\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f7fa3601",
   "metadata": {},
   "source": [
    "We can use the `foreachBatch()` function in the writestream statement to apply the above function to each batch of the customer bureau stream."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "aa6de855",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 24:====================================================> (196 + 2) / 200]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Current batch is empty! Skipping..\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Epoch ID: 1, Timestamp: 2022-10-18 14:56:00,  Data: [{\"State\":\"WEST BENGAL\",\"target_customers\":494},{\"State\":\"CHATTISGARH\",\"target_customers\":434},{\"State\":\"RAJASTHAN\",\"target_customers\":294},{\"State\":\"GUJARAT\",\"target_customers\":382},{\"State\":\"KARNATAKA\",\"target_customers\":62},{\"State\":\"ORISSA\",\"target_customers\":521},{\"State\":\"UTTAR PRADESH\",\"target_customers\":170},{\"State\":\"MADHYA PRADESH\",\"target_customers\":4161},{\"State\":\"MAHARASHTRA\",\"target_customers\":34},{\"State\":\"PUNJAB\",\"target_customers\":564},{\"State\":\"UTTARAKHAND\",\"target_customers\":288},{\"State\":\"HARYANA\",\"target_customers\":252},{\"State\":\"TELANGANA\",\"target_customers\":15},{\"State\":\"ANDHRA PRADESH\",\"target_customers\":454}]\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 48:===========>                                           (40 + 2) / 200]\r"
     ]
    }
   ],
   "source": [
    "# Starting the producer to send the count of predicted top-up customers\n",
    "query = cust_bureau_df.writeStream.foreachBatch(prediction_stream).outputMode(\"append\").start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "id": "ff7feae9",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Epoch ID: 2, Timestamp: 2022-10-18 13:32:00,  Data: [{\"State\":\"CHATTISGARH\",\"target_customers\":3},{\"State\":\"ORISSA\",\"target_customers\":12},{\"State\":\"MADHYA PRADESH\",\"target_customers\":4}]\n"
     ]
    }
   ],
   "source": [
    "# Stopping the query\n",
    "query.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c69afb4d",
   "metadata": {},
   "source": [
    "**NOTE**: The above output is incomplete on the producer side and consumer side because the cloud VM at Monash University is unable to take the load of this operation and generate the map. I have already tested the logic of the code producing the map on my local system and so I am confident that it will not fail if this entire application from step 1 to step 3 is run on a system with a better memory.\n",
    "\n",
    "I strongly **recommend** using a system with good memory to execute the code for the 3 steps especially for section 2.8 and Step 3: Consumer."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f7b272d3",
   "metadata": {},
   "source": [
    "## References\n",
    "\n",
    "* Parquet Files - Spark 3.1.2 Documentation. (n.d.). Spark.apache.org. https://spark.apache.org/docs/latest/sql-data-sources-parquet.html"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.16"
  },
  "latex_envs": {
   "LaTeX_envs_menu_present": true,
   "autoclose": false,
   "autocomplete": true,
   "bibliofile": "biblio.bib",
   "cite_by": "apalike",
   "current_citInitial": 1,
   "eqLabelWithNumbers": true,
   "eqNumInitial": 1,
   "hotkeys": {
    "equation": "Ctrl-E",
    "itemize": "Ctrl-I"
   },
   "labels_anchors": false,
   "latex_user_defs": false,
   "report_style_numbering": false,
   "user_envs_cfg": false
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}