spark_twit.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark import SparkConf,SparkContext\n",
    "from pyspark.streaming import StreamingContext\n",
    "from pyspark.sql import Row,SQLContext\n",
    "from pyspark.sql import SparkSession\n",
    "import sys\n",
    "import requests\n",
    "import socket\n",
    "import json\n",
    "import traceback\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def send_df_to_csv(df):\n",
    "# extract the hashtags from dataframe and convert them into array\n",
    "    top_tags = [str(t.hashtag) for t in df.select(\"hashtag\").collect()]\n",
    "# extract the counts from dataframe and convert them into array\n",
    "    tags_count = [p.hashtag_count for p in df.select(\"hashtag_count\").collect()]\n",
    "    #path=\n",
    "    url = ''\n",
    "    request_data = {'label': str(top_tags), 'data': str(tags_count)}\n",
    "    response = requests.post(url, data=request_data)\n",
    "    df.toPandas().to_csv(r'\\twit_file.csv', header=True, mode='a', index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_sql_context_instance(spark_context):\n",
    "    if ('sqlContextSingletonInstance' not in globals()):\n",
    "        globals()['sqlContextSingletonInstance'] = SQLContext(spark_context)\n",
    "    return globals()['sqlContextSingletonInstance']\n",
    "def process_rdd(time, rdd):\n",
    "    print(\"----------- %s -----------\" % str(time))\n",
    "    try:\n",
    "        # Get spark sql singleton context from the current context\n",
    "        sql_context = get_sql_context_instance(rdd.context)\n",
    "        # convert the RDD to Row RDD\n",
    "        row_rdd = rdd.map(lambda w: Row(hashtag=w[0], hashtag_count=w[1]))\n",
    "        # create a DF from the Row RDD\n",
    "        df_hash = sql_context.createDataFrame(row_rdd)\n",
    "        # Register the dataframe as table\n",
    "        df_hash.registerTempTable(\"hashtags\")\n",
    "        # get the top 10 hashtags from the table using SQL and print them\n",
    "        hashtag_counts_df = sql_context.sql(\"select hashtag, hashtag_count from hashtags order by hashtag_count desc limit 10\")\n",
    "        hashtag_counts_df.show()\n",
    "        # call this method to prepare top 10 hashtags DF and send them\n",
    "        send_df_to_csv(hashtag_counts_df)\n",
    "    except:\n",
    "        print(traceback.format_exc())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create spark configuration\n",
    "conf = SparkConf()\n",
    "conf.setAppName(\"TwitterStrimApp\")\n",
    "# create spark context with the above configuration\n",
    "sc = SparkContext(conf=conf)\n",
    "sc.setLogLevel(\"ERROR\")\n",
    "# create the Streaming Context from the above spark context with interval size 60 seconds\n",
    "ssc = StreamingContext(sc, 720)\n",
    "# setting a checkpoint to allow RDD recovery\n",
    "ssc.checkpoint(\"checkpoint_tweet\")\n",
    "dataStream = ssc.socketTextStream(\"localhost\",9009)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# split each tweet into words\n",
    "words = dataStream.flatMap(lambda line: line.split(\" \"))\n",
    "# filter the words to get only hashtags, then map each hashtag to be a pair of (hashtag,1)\n",
    "hashtags = words.filter(lambda w: '#' in w).map(lambda x: (x, 1))\n",
    "# adding the count of each hashtag to its last count\n",
    "total_tags = hashtags.reduceByKey(lambda a, b: a + b)\n",
    "# do processing for each RDD generated in each interval\n",
    "total_tags.foreachRDD(process_rdd)\n",
    "# start the streaming computation\n",
    "ssc.start()\n",
    "# wait for the streaming to finish\n",
    "ssc.awaitTerminationOrTimeout(3000) "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}