Spark Streaming.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "0d427efe",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "from pyspark import SparkContext\n",
    "from pyspark.streaming import StreamingContext\n",
    "from collections import namedtuple\n",
    "from pyspark.sql.functions import desc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "c89cd1a5",
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "c:\\Users\\ADMIN\\AppData\\Local\\Programs\\Python\\Python312\\Lib\\site-packages\\pyspark\\streaming\\context.py:72: FutureWarning: DStream is deprecated as of Spark 3.4.0. Migrate to Structured Streaming.\n",
      "  warnings.warn(\n",
      "c:\\Users\\ADMIN\\AppData\\Local\\Programs\\Python\\Python312\\Lib\\site-packages\\pyspark\\sql\\context.py:113: FutureWarning: Deprecated in 3.0.0. Use SparkSession.builder.getOrCreate() instead.\n",
      "  warnings.warn(\n"
     ]
    }
   ],
   "source": [
    "ssc = StreamingContext(sc,10)\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "socket_stream = ssc.socketTextStream(\"localhost\",9999)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "lines = socket_stream.window( 20 )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "c7c19d3d",
   "metadata": {},
   "outputs": [],
   "source": [
    "fields = (\"tag\", \"count\")\n",
    "Tweet = namedtuple(\"Tweet\",fields)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Use Parenthesis for multiple lines or use \\.\n",
    "( lines.flatMap( lambda text: text.split( \" \" ) ) #Splits to a list\n",
    "  .filter( lambda word: word.lower().startswith(\"#\") ) # Checks for hashtag calls\n",
    "  .map( lambda word: ( word.lower(), 1 ) ) # Lower cases the word\n",
    "  .reduceByKey( lambda a, b: a + b ) # Reduces\n",
    "  .map( lambda rec: Tweet( rec[0], rec[1] ) ) # Stores in a Tweet Object\n",
    "  .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") ) # Sorts Them in a DF\n",
    "  .limit(10).registerTempTable(\"tweets\") ) ) # Registers to a table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "ec1e0d18",
   "metadata": {},
   "outputs": [],
   "source": [
    "ssc.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "8d4a0e9f",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import explode, split, col, desc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "6824a81d",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"TwitterHashtagPopularity\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "123a6931",
   "metadata": {},
   "outputs": [],
   "source": [
    "# read the tweet data from socket\n",
    "tweet_df = spark \\\n",
    "    .readStream \\\n",
    "    .format(\"socket\") \\\n",
    "    .option(\"host\", \"127.0.0.1\") \\\n",
    "    .option(\"port\", 9999) \\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "ed99a691",
   "metadata": {},
   "outputs": [],
   "source": [
    "# type cast the column value\n",
    "tweet_df = tweet_df.selectExpr(\"CAST(value AS STRING)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "ab1bad2c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# split words based on space, filter out hashtag values and group them up\n",
    "tweets_tab = tweet_df.withColumn('word', explode(split(col('value'), ' '))) \\\n",
    "    .groupBy('word') \\\n",
    "    .count() \\\n",
    "    .sort('count', ascending=False) \\\n",
    "    .filter(col('word').contains('#'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "0dad6385",
   "metadata": {},
   "outputs": [],
   "source": [
    "# write the above data into memory. consider the entire analysis in all iteration (output mode = complete). and let the trigger runs in every 2 secs.\n",
    "writeTweet = tweets_tab.writeStream. \\\n",
    "    outputMode(\"complete\"). \\\n",
    "    format(\"console\"). \\\n",
    "    queryName(\"tweetquery\"). \\\n",
    "    trigger(processingTime='2 seconds'). \\\n",
    "    start(). \\\n",
    "    awaitTermination()"
   ]
  }
 ],
 "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.12.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}