transforming_and_querying_the_streaming_data.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark functions\n",
    "from pyspark.sql.functions import *\n",
    "# URL processing\n",
    "import urllib"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Specify file type to be csv\n",
    "file_type = \"csv\"\n",
    "# Indicates file has first row as the header\n",
    "first_row_is_header = \"true\"\n",
    "# Indicates file has comma as the delimeter\n",
    "delimiter = \",\"\n",
    "# Read the CSV file to spark dataframe\n",
    "aws_keys_df = spark.read.format(file_type)\\\n",
    ".option(\"header\", first_row_is_header)\\\n",
    ".option(\"sep\", delimiter)\\\n",
    ".load(\"/FileStore/tables/authentication_credentials.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the AWS access key and secret key from the spark dataframe\n",
    "ACCESS_KEY = aws_keys_df.where(col('User name')=='databricks-user').select('Access key ID').collect()[0]['Access key ID']\n",
    "SECRET_KEY = aws_keys_df.where(col('User name')=='databricks-user').select('Secret access key').collect()[0]['Secret access key']\n",
    "# Encode the secrete key\n",
    "ENCODED_SECRET_KEY = urllib.parse.quote(string=SECRET_KEY, safe=\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import from_json, col, base64\n",
    "from pyspark.sql.types import StructType, StructField, IntegerType, TimestampType, FloatType, StringType\n",
    "from pyspark.streaming.kinesis import KinesisUtils, InitialPositionInStream\n",
    "\n",
    "kinesisStreamName = \"streaming-0e2b04098249-pin\"\n",
    "regionName = \"us-east-1\"\n",
    "\n",
    "# Read data from Kinesis stream using structured streaming\n",
    "streaming_df_pin_kinesis = spark.readStream \\\n",
    "    .format(\"kinesis\") \\\n",
    "    .option(\"streamName\", kinesisStreamName) \\\n",
    "    .option(\"regionName\", regionName) \\\n",
    "    .option(\"initialPosition\", \"TRIM_HORIZON\") \\\n",
    "    .option(\"format\", \"json\") \\\n",
    "    .option(\"awsAccessKey\", ACCESS_KEY) \\\n",
    "    .option(\"awsSecretKey\", SECRET_KEY) \\\n",
    "    .option(\"inferSchema\", \"true\") \\\n",
    "    .load()\n",
    "\n",
    "# define the schema for the table (must use the correct column names)\n",
    "schema = StructType([\n",
    "    StructField(\"index\", IntegerType()),\n",
    "    StructField(\"unique_id\", StringType()),\n",
    "    StructField(\"title\", StringType()),\n",
    "    StructField(\"description\", StringType()),\n",
    "    StructField(\"poster_name\", StringType()),\n",
    "    StructField(\"follower_count\", StringType()),\n",
    "    StructField(\"tag_list\", StringType()),\n",
    "    StructField(\"is_image_or_video\", StringType()),\n",
    "    StructField(\"image_src\", StringType()),\n",
    "    StructField(\"downloaded\", IntegerType()),\n",
    "    StructField(\"save_location\", StringType()),\n",
    "    StructField(\"category\", StringType())\n",
    "])\n",
    "\n",
    "# Define a UDF to deserialize the data column\n",
    "deserialise_udf_pin = from_json(col(\"data\").cast(\"string\"), schema)\n",
    "\n",
    "# Apply the UDF and select the desired columns\n",
    "streaming_df_pin = streaming_df_pin_kinesis.withColumn(\"cast_data_pin\", deserialise_udf_pin) \\\n",
    "    .select(\"cast_data_pin.index\", \"cast_data_pin.unique_id\", \"cast_data_pin.title\", \"cast_data_pin.description\", \"cast_data_pin.poster_name\", \"cast_data_pin.follower_count\", \"cast_data_pin.tag_list\", \"cast_data_pin.is_image_or_video\", \"cast_data_pin.image_src\", \"cast_data_pin.downloaded\", \"cast_data_pin.save_location\", \"cast_data_pin.category\")\n",
    "\n",
    "display(streaming_df_pin)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "kinesisStreamName = \"streaming-0e2b04098249-geo\"\n",
    "regionName = \"us-east-1\"\n",
    "\n",
    "# Read data from Kinesis stream using structured streaming\n",
    "streaming_df_geo_kinesis = spark.readStream \\\n",
    "    .format(\"kinesis\") \\\n",
    "    .option(\"streamName\", kinesisStreamName) \\\n",
    "    .option(\"regionName\", regionName) \\\n",
    "    .option(\"initialPosition\", \"TRIM_HORIZON\") \\\n",
    "    .option(\"format\", \"json\") \\\n",
    "    .option(\"awsAccessKey\", ACCESS_KEY) \\\n",
    "    .option(\"awsSecretKey\", SECRET_KEY) \\\n",
    "    .option(\"inferSchema\", \"true\") \\\n",
    "    .load()\n",
    "\n",
    "\n",
    "# Define the schema for the table\n",
    "schema = StructType([\n",
    "    StructField(\"ind\", IntegerType()),\n",
    "    StructField(\"timestamp\", TimestampType()),\n",
    "    StructField(\"latitude\", FloatType()),\n",
    "    StructField(\"longitude\", FloatType()),\n",
    "    StructField(\"country\", StringType())\n",
    "])\n",
    "\n",
    "# Define a UDF to deserialize the data column\n",
    "deserialise_udf_geo = from_json(col(\"data\").cast(\"string\"), schema)\n",
    "\n",
    "# Apply the UDF and select the desired columns\n",
    "streaming_df_geo = streaming_df_geo_kinesis.withColumn(\"cast_data_geo\", deserialise_udf_geo) \\\n",
    "    .select(\"cast_data_geo.ind\", \"cast_data_geo.timestamp\", \"cast_data_geo.latitude\", \"cast_data_geo.longitude\", \"cast_data_geo.country\")\n",
    "\n",
    "display(streaming_df_geo)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "kinesisStreamName = \"streaming-0e2b04098249-user\"\n",
    "regionName = \"us-east-1\"\n",
    "\n",
    "# Read data from Kinesis stream using structured streaming\n",
    "streaming_df_user_kinesis = spark.readStream \\\n",
    "    .format(\"kinesis\") \\\n",
    "    .option(\"streamName\", kinesisStreamName) \\\n",
    "    .option(\"regionName\", regionName) \\\n",
    "    .option(\"initialPosition\", \"TRIM_HORIZON\") \\\n",
    "    .option(\"format\", \"json\") \\\n",
    "    .option(\"awsAccessKey\", ACCESS_KEY) \\\n",
    "    .option(\"awsSecretKey\", SECRET_KEY) \\\n",
    "    .option(\"inferSchema\", \"true\") \\\n",
    "    .load()\n",
    "\n",
    "\n",
    "# Define the schema for the table\n",
    "schema = StructType([\n",
    "    StructField(\"ind\", IntegerType()),\n",
    "    StructField(\"first_name\", StringType()),\n",
    "    StructField(\"last_name\", StringType()),\n",
    "    StructField(\"age\", StringType()),\n",
    "    StructField(\"date_joined\", TimestampType())\n",
    "])\n",
    "\n",
    "# Define a UDF to deserialize the data column\n",
    "deserialise_udf_user = from_json(col(\"data\").cast(\"string\"), schema)\n",
    "\n",
    "# Apply the UDF and select the desired columns\n",
    "streaming_df_user = streaming_df_user_kinesis.withColumn(\"cast_data_user\", deserialise_udf_user) \\\n",
    "    .select(\"cast_data_user.ind\", \"cast_data_user.first_name\", \"cast_data_user.last_name\", \"cast_data_user.age\", \"cast_data_user.date_joined\")\n",
    "\n",
    "display(streaming_df_user)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transforming and cleaning the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import regexp_replace, col\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\n",
    "    \"description\",\n",
    "    when(\n",
    "        (streaming_df_pin.description == \"Untitled\") |\n",
    "        (streaming_df_pin.description == \"No description available\") |\n",
    "        (streaming_df_pin.description == \"No description available Story format\"),\n",
    "        \"None\"\n",
    "    ).otherwise(streaming_df_pin.description)\n",
    ")\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\n",
    "    \"tag_list\",\n",
    "    when(\n",
    "        (streaming_df_pin.tag_list == \"N,o, ,T,a,g,s, ,A,v,a,i,l,a,b,l,e\")|\n",
    "        (streaming_df_pin.tag_list == \"0\"),\n",
    "        \"None\"\n",
    "    ).otherwise(streaming_df_pin.tag_list)\n",
    ")\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\n",
    "    \"title\",\n",
    "    when(\n",
    "        (streaming_df_pin.title == \"No Title Data Available\"),\n",
    "        \"None\"\n",
    "    ).otherwise(streaming_df_pin.title)\n",
    ")\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\n",
    "    \"follower_count\",\n",
    "    when(\n",
    "        (streaming_df_pin.follower_count == \"User Info Error\"),\n",
    "        \"0\"\n",
    "    ).otherwise(streaming_df_pin.follower_count)\n",
    ")\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\n",
    "    \"image_src\",\n",
    "    when(\n",
    "        (streaming_df_pin.image_src == \"Image src error.\"),\n",
    "        \"None\"\n",
    "    ).otherwise(streaming_df_pin.image_src)\n",
    ")\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\n",
    "    \"poster_name\",\n",
    "    when(\n",
    "        (streaming_df_pin.poster_name == \"User Info Error\"),\n",
    "        \"None\"\n",
    "    ).otherwise(streaming_df_pin.poster_name)\n",
    ")\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\n",
    "    \"downloaded\",\n",
    "    when(\n",
    "        (streaming_df_pin.downloaded == \"None\"),\n",
    "        \"0\"\n",
    "    ).otherwise(streaming_df_pin.downloaded)\n",
    ")\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\"follower_count\", regexp_replace(\"follower_count\", \"k\", \"000\"))\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\"follower_count\", regexp_replace(\"follower_count\", \"M\", \"000000\"))\n",
    "\n",
    "# change the datatype of the \"follower_count\" column to int\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\"follower_count\", col(\"follower_count\").cast(\"int\"))\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumnRenamed(\"index\", \"ind\")\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.withColumn(\"save_location\", regexp_replace(\"save_location\", \"Local save in \", \"\"))\n",
    "\n",
    "streaming_df_pin = streaming_df_pin.select(col(\"ind\"), col(\"unique_id\"), col(\"title\"), col(\"description\"), col(\"follower_count\"), col(\"poster_name\"), col(\"tag_list\"), col(\"is_image_or_video\"), col(\"image_src\"), col(\"save_location\"), col(\"category\"))\n",
    "\n",
    "display(streaming_df_pin)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import array\n",
    "\n",
    "streaming_df_geo = streaming_df_geo.withColumn(\"coordinates\", array(\"latitude\", \"longitude\"))\n",
    "streaming_df_geo = streaming_df_geo.drop(\"latitude\", \"longitude\")\n",
    "\n",
    "# change the datatype of the \"timestamp\" column to timestamp\n",
    "streaming_df_geo = streaming_df_geo.withColumn(\"timestamp\", col(\"timestamp\").cast(\"timestamp\"))\n",
    "\n",
    "streaming_df_geo = streaming_df_geo.select(col(\"ind\"), col(\"country\"), col(\"coordinates\"), col(\"timestamp\"))\n",
    "\n",
    "display(streaming_df_geo)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import concat\n",
    "\n",
    "streaming_df_user = streaming_df_user.withColumn('user_name', concat(streaming_df_user.first_name, lit(' '), streaming_df_user.last_name))\n",
    "\n",
    "streaming_df_user = streaming_df_user.drop(\"first_name\", \"last_name\")\n",
    "\n",
    "# change the datatype of the \"date_joined\" column to timestamp\n",
    "streaming_df_user = streaming_df_user.withColumn(\"date_joined\", col(\"date_joined\").cast(\"timestamp\"))\n",
    "\n",
    "streaming_df_user = streaming_df_user.select(col(\"ind\"), col(\"user_name\"), col(\"age\"), col(\"date_joined\"))\n",
    "\n",
    "display(streaming_df_user)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Save the data to Delta Tables on Databricks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write the streaming DataFrame to a temporary table\n",
    "streaming_df_pin.writeStream.format(\"memory\").queryName(\"pin_temp_table\").start()\n",
    "\n",
    "# Read the temporary table as a non-streaming DataFrame\n",
    "not_streaming_df_pin = spark.sql(\"SELECT * FROM pin_temp_table\")\n",
    "\n",
    "# Display the non-streaming DataFrame\n",
    "display(not_streaming_df_pin)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write the streaming DataFrame to a temporary table\n",
    "streaming_df_geo.writeStream.format(\"memory\").queryName(\"geo_temp_table\").start()\n",
    "\n",
    "# Read the temporary table as a non-streaming DataFrame\n",
    "not_streaming_df_geo = spark.sql(\"SELECT * FROM geo_temp_table\")\n",
    "\n",
    "# Display the non-streaming DataFrame\n",
    "display(not_streaming_df_geo)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write the streaming DataFrame to a temporary table\n",
    "streaming_df_user.writeStream.format(\"memory\").queryName(\"user_temp_table\").start()\n",
    "\n",
    "# Read the temporary table as a non-streaming DataFrame\n",
    "not_streaming_df_user = spark.sql(\"SELECT * FROM user_temp_table\")\n",
    "\n",
    "# Display the non-streaming DataFrame\n",
    "display(not_streaming_df_user)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Save the DataFrame's as Delta tables\n",
    "not_streaming_df_pin.write.format(\"delta\").mode(\"overwrite\").saveAsTable(\"0e2b04098249_pin_table\")\n",
    "\n",
    "not_streaming_df_geo.write.format(\"delta\").mode(\"overwrite\").saveAsTable(\"0e2b04098249_geo_table\")\n",
    "\n",
    "not_streaming_df_user.write.format(\"delta\").mode(\"overwrite\").saveAsTable(\"0e2b04098249_user_table\")"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}