Kafka-Pyspark-Integration.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "ef7cc06d-1d9e-48c1-8f71-b8eaf6ff1079",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import from_json, col ,regexp_replace, substring\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.types import StructType, StructField, StringType, IntegerType,DoubleType,FloatType,DateType\n",
    "from pyspark.sql.functions import expr\n",
    "import seaborn as sns\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "scala_version = '2.12'  \n",
    "spark_version = '3.2.1'\n",
    "packages = [\n",
    "    f'org.apache.spark:spark-sql-kafka-0-10_{scala_version}:{spark_version}',\n",
    "    f'org.apache.spark:spark-streaming-kafka-0-10_{scala_version}:{spark_version}',\n",
    "    f'org.apache.spark:spark-token-provider-kafka-0-10_{scala_version}:{spark_version}',\n",
    "    'org.apache.kafka:kafka-clients:2.1.1',\n",
    "    'org.apache.commons:commons-pool2:2.8.0'\n",
    "]\n",
    "\n",
    "# Initialize a SparkSession object locally\n",
    "\n",
    "spark = SparkSession.builder\\\n",
    "   .master(\"local\")\\\n",
    "   .appName(\"kafka-example\")\\\n",
    "   .config(\"spark.jars.packages\", \",\".join(packages))\\\n",
    "   .getOrCreate()\n",
    "spark.sparkContext.setLogLevel('ERROR')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "3f0e2beb-e923-42e9-88ab-e2b56960dc6e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Set up a streaming DataFrame 'dfraw'\n",
    "\n",
    "brokers = \"Public DNS of EC2:9092\"\n",
    "topic = \"mytopic1\"\n",
    "\n",
    "dfraw = spark \\\n",
    "  .readStream \\\n",
    "  .format(\"kafka\") \\\n",
    "  .option(\"kafka.bootstrap.servers\", brokers) \\\n",
    "  .option(\"subscribe\", topic) \\\n",
    "  .option(\"includeHeaders\",\"true\") \\\n",
    "  .option(\"startingOffsets\",\"latest\") \\\n",
    "  .load() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "9cede1db-e3d7-4782-81a8-26446ceb7186",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Set up a streaming query \n",
    "\n",
    "rawQuery = dfraw \\\n",
    "        .writeStream \\\n",
    "        .queryName(\"query_data\")\\\n",
    "        .format(\"memory\")\\\n",
    "        .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "235f5110-4193-4da9-914a-335d25707912",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "kafka_df = spark.sql(\"select CAST(value AS STRING), topic, timestamp from query_data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "f4fef928-05f3-470b-b501-ce13929a20f3",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 26:>                                                         (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------+-----------------------+\n",
      "|value                                                                                                                                                                                                                        |topic   |timestamp              |\n",
      "+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------+-----------------------+\n",
      "|[{\"user_id\": 89, \"user_sex\": \"male\", \"user_age\": 33, \"user_country\": \"Finland\", \"rating\": 4.5, \"comment\": \"\\\"Albus Dumbledore's presence adds gravitas.\\\"\", \"favourite_character\": \"Albus Dumbledore\", \"date\": \"2005-04-25\"}]|mytopic1|2024-02-09 09:02:35.933|\n",
      "+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------+-----------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "kafka_df.show(1, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "9e35edb8-93cc-4fd4-b3d7-e3e1cd0298b6",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------+-----------------------+-------------------+\n",
      "|value                                                                                                                                                                                                                        |topic   |timestamp              |favourite_character|\n",
      "+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------+-----------------------+-------------------+\n",
      "|[{\"user_id\": 89, \"user_sex\": \"male\", \"user_age\": 33, \"user_country\": \"Finland\", \"rating\": 4.5, \"comment\": \"\\\"Albus Dumbledore's presence adds gravitas.\\\"\", \"favourite_character\": \"Albus Dumbledore\", \"date\": \"2005-04-25\"}]|mytopic1|2024-02-09 09:02:35.933|Albus Dumbledore   |\n",
      "+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------+-----------------------+-------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "kafka_df = kafka_df.withColumn(\n",
    "    \"favourite_character\",\n",
    "    expr(\"get_json_object(value, '$[0].favourite_character')\"))\n",
    "\n",
    "kafka_df.show(1,truncate=False)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "cbf5f732-1397-4f0b-b628-69a8ce03e144",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 76:>                                                         (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----+\n",
      "|favourite_character|count|\n",
      "+-------------------+-----+\n",
      "|Ron Weasley        |9    |\n",
      "|Severus Snape      |3    |\n",
      "|Hermione Granger   |10   |\n",
      "|Harry Potter       |12   |\n",
      "|Albus Dumbledore   |10   |\n",
      "|Rubeus Hagrid      |8    |\n",
      "|Neville Longbottom |6    |\n",
      "|Draco Malfoy       |5    |\n",
      "+-------------------+-----+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "kafka_df.groupBy('favourite_character').count().show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7f6c7a88-a58e-4ba6-afa8-fbab56fb7c17",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}