05 Sentiment Analysis Stream.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "4bc1f4d7",
   "metadata": {},
   "source": [
    "# Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "217786dc",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkContext\n",
    "from pyspark.streaming import StreamingContext\n",
    "import pyspark.sql.functions as F\n",
    "\n",
    "from pyspark.sql.types import MapType, StringType, IntegerType, ArrayType, StructType, \\\n",
    "                              StructField, LongType, DoubleType, BooleanType, FloatType\n",
    "\n",
    "from IPython.display import display, clear_output\n",
    "import time\n",
    "\n",
    "\n",
    "# for the model\n",
    "from pyspark.ml.classification import NaiveBayes\n",
    "from pyspark.ml.feature import CountVectorizer\n",
    "from pyspark.ml.feature import Tokenizer, RegexTokenizer\n",
    "from pyspark.ml.feature import StopWordsRemover\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.ml.feature import NGram\n",
    "\n",
    "from pyspark.ml import Pipeline, PipelineModel\n",
    "\n",
    "from pyspark.ml.classification import NaiveBayes\n",
    "\n",
    "from pyspark.ml.feature import Word2Vec\n",
    "from pyspark.sql import Row\n",
    "from json import dumps\n",
    "from pyspark.sql import udf\n",
    "import json\n",
    "\n",
    "import cld3\n",
    "\n",
    "import os\n",
    "import shutil\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b0092947",
   "metadata": {},
   "source": [
    "# Step 1: Building Spark streaming and Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "54e56a74",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating Spark session\n",
    "spark = SparkSession.builder \\\n",
    "        .appName('kafka') \\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "6fe91bf6",
   "metadata": {},
   "outputs": [],
   "source": [
    "## To always show the results of DataFrames and improve the formatting of the output\n",
    "spark.conf.set(\"spark.sql.repl.eagerEval.enabled\",True)\n",
    "\n",
    "## To allow future conversion of Spark DataFrame into Pandas DataFrame\n",
    "spark.conf.set(\"spark.sql.execution.arrow.enabled\", True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "5cddedce",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read data from the topic \n",
    "df = spark \\\n",
    "  .readStream \\\n",
    "  .format(\"kafka\") \\\n",
    "  .option(\"kafka.bootstrap.servers\", \"broker:29092\") \\\n",
    "  .option(\"subscribe\", \"engtweets\") \\\n",
    "  .option(\"startingOffsets\",\"earliest\") \\\n",
    "  .option(\"includeHeaders\", \"true\") \\\n",
    "  .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "a2962838",
   "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",
      " |-- headers: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- key: string (nullable = true)\n",
      " |    |    |-- value: binary (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Print the Schema\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "6cf72de6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Schema of the english tweets data\n",
    "\n",
    "schema_tweet = StructType([\n",
    "    StructField(\"CREATEDAT\", LongType(),  True),\n",
    "    StructField(\"ID\", LongType(),  True),\n",
    "    StructField(\"USER_ID\", LongType(),  True),\n",
    "    StructField(\"USER_NAME\", StringType(),  True),\n",
    "    StructField(\"SCREENNAME\", StringType(),  True),\n",
    "    StructField(\"USER_LOCATION\", StringType(), True),\n",
    "    StructField(\"FOLLOW_COUNT\", IntegerType(),  True),\n",
    "    StructField(\"FRIEND_COUNT\", IntegerType(),  True),\n",
    "    StructField(\"USER_CREAT_AT\", LongType(),  True),\n",
    "    StructField(\"USER_FAV_COUNT\", IntegerType(), True),\n",
    "    StructField(\"USER_VERIFIED\", BooleanType(), True),\n",
    "    StructField(\"USER_LANG\", StringType(),  True),\n",
    "    StructField(\"TEXT\", StringType(),  True),\n",
    "    StructField(\"LANG\", StringType(),  True),\n",
    "    StructField(\"GAO_LAT\", DoubleType(), True),\n",
    "    StructField(\"GEO_LONG\", DoubleType(),  True),\n",
    "    StructField(\"PLACE_NAME\", StringType(),  True),\n",
    "    StructField(\"PLACE_COUNTRY\", StringType(), True),\n",
    "    StructField(\"HASHTAG\", ArrayType( StringType()), True),\n",
    "    StructField(\"USER_MENTION_NAME\", ArrayType( StringType()), True)\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "31d1e51b",
   "metadata": {},
   "source": [
    "# Step 2: Cleaning tweets text with the same process of training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "39f218e3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Transform data of the topic\n",
    "clean_df = df.selectExpr(\"CAST(key AS STRING)\", \"CAST(value AS STRING)\")\\\n",
    "    .withColumn(\"value\", F.from_json(\"value\", schema_tweet)) \\\n",
    "    .select(\"key\", F.col('value.*'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "480969a2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: string (nullable = true)\n",
      " |-- CREATEDAT: long (nullable = true)\n",
      " |-- ID: long (nullable = true)\n",
      " |-- USER_ID: long (nullable = true)\n",
      " |-- USER_NAME: string (nullable = true)\n",
      " |-- SCREENNAME: string (nullable = true)\n",
      " |-- USER_LOCATION: string (nullable = true)\n",
      " |-- FOLLOW_COUNT: integer (nullable = true)\n",
      " |-- FRIEND_COUNT: integer (nullable = true)\n",
      " |-- USER_CREAT_AT: long (nullable = true)\n",
      " |-- USER_FAV_COUNT: integer (nullable = true)\n",
      " |-- USER_VERIFIED: boolean (nullable = true)\n",
      " |-- USER_LANG: string (nullable = true)\n",
      " |-- TEXT: string (nullable = true)\n",
      " |-- LANG: string (nullable = true)\n",
      " |-- GAO_LAT: double (nullable = true)\n",
      " |-- GEO_LONG: double (nullable = true)\n",
      " |-- PLACE_NAME: string (nullable = true)\n",
      " |-- PLACE_COUNTRY: string (nullable = true)\n",
      " |-- HASHTAG: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- USER_MENTION_NAME: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "clean_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8b914038",
   "metadata": {},
   "source": [
    "Remove words that start with a character"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "8028df1f",
   "metadata": {},
   "outputs": [],
   "source": [
    "#remove words that start with a character\n",
    "def remove_start(text, start_chr):\n",
    "    return \" \".join(filter(lambda word:word[0]!=start_chr, text.split()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "d3f5a9b1",
   "metadata": {},
   "outputs": [],
   "source": [
    "UDF_remove_start = F.udf(lambda text, start_chr: remove_start(text, start_chr), \n",
    "                        StringType())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c38f7135",
   "metadata": {},
   "source": [
    "Detect language"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "46baaa04",
   "metadata": {},
   "outputs": [],
   "source": [
    "#detect language: english\n",
    "def detect_lang(text):\n",
    "    try:\n",
    "        return cld3.get_language(text)[0]\n",
    "    except:\n",
    "        return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "b97b0210",
   "metadata": {},
   "outputs": [],
   "source": [
    "UDF_detect_lang = F.udf(lambda text: detect_lang(text), \n",
    "                        StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "102b766d",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Apply the above functions to the dataframe\n",
    "# 1) Reading tweets\n",
    "# 2) Removing words that strat with @\n",
    "# 3) Removing words that strat with #\n",
    "# 4) To lowercase\n",
    "# 5) Add language column\n",
    "# 6) Filter only enlish\n",
    "# 7) Remove lang column\n",
    "# 8) Remove numbers\n",
    "# 9) Remove symbols\n",
    "# 10) Keep obs with at least one letter\n",
    "\n",
    "final_df = clean_df.withColumn('new_text',UDF_remove_start(F.col('TEXT'), F.lit('@'))) \\\n",
    "    .withColumn('new_text',UDF_remove_start(F.col('TEXT'), F.lit('#'))) \\\n",
    "    .withColumn('new_text', F.lower(F.col('TEXT'))) \\\n",
    "    .withColumn('LANG', UDF_detect_lang(F.col('TEXT'))) \\\n",
    "    .filter(F.col('LANG') == 'en') \\\n",
    "    .withColumn(\"new_text\", F.regexp_replace(F.col(\"TEXT\"), r'[0-9]', '')) \\\n",
    "    .withColumn(\"new_text\", F.regexp_replace(F.col(\"TEXT\"), r'[$-/:-?{-~!\"^_`\\[\\]]', '')) \\\n",
    "    .filter(F.col('new_text').rlike(\"^.*[a-zA-Z]+.*$\")) \\\n",
    "    .select(F.col('key'),F.col('USER_NAME'),F.col('text').alias(\"raw_text\"),F.col('new_text').alias(\"text\")) #F.col('id'), ,"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "8549fe53",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: string (nullable = true)\n",
      " |-- USER_NAME: string (nullable = true)\n",
      " |-- raw_text: string (nullable = true)\n",
      " |-- text: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "final_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "adc8a4fd",
   "metadata": {},
   "source": [
    "# Step 3: Predictions "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "f2513f61",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Load the model\n",
    "pipelineModel = PipelineModel.load(\"models/naivebayes.model\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "4bd56ea8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Getting predictions\n",
    "pred = pipelineModel.transform(final_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "3d224dbc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: string (nullable = true)\n",
      " |-- USER_NAME: string (nullable = true)\n",
      " |-- raw_text: string (nullable = true)\n",
      " |-- text: string (nullable = true)\n",
      " |-- tokens: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- ngrams: array (nullable = true)\n",
      " |    |-- element: string (containsNull = 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 prediction schema\n",
    "pred.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "da69f8e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Select some columns\n",
    "prediction= pred.select(F.col('key'),F.col('USER_NAME'),F.col('raw_text').alias(\"tweet_text\"),F.col('prediction').alias(\"sentiment\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "7775c0da",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Changing 0 to negative and 1 to positive\n",
    "prediction = prediction.withColumn(\"sentiment\", \n",
    "                                         F.when((F.col(\"sentiment\")== 0), \"negative\")\n",
    "                                          .otherwise(\"positive\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "7c802104",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: string (nullable = true)\n",
      " |-- USER_NAME: string (nullable = true)\n",
      " |-- tweet_text: string (nullable = true)\n",
      " |-- sentiment: string (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Print the schema\n",
    "prediction.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "43af3657",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Transform the columns to json format to upload to kafka (required format)\n",
    "\n",
    "prediction = prediction.withColumn(\"value\",F.create_map(\n",
    "                                        F.lit(\"USER_NAME\"),F.col(\"USER_NAME\"),\n",
    "                                        F.lit(\"tweet_text\"),F.col(\"tweet_text\"),\n",
    "                                        F.lit(\"sentiment\"),F.col(\"sentiment\")\n",
    "                                    )) \\\n",
    "    .drop(\"USER_NAME\",\"tweet_text\", \"sentiment\") \\\n",
    "    .select(F.col(\"key\").cast(StringType()).alias(\"key\"), F.to_json(F.col(\"value\")).alias(\"value\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "aaa92d1c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: string (nullable = true)\n",
      " |-- value: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "prediction.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "add7ada1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Remove checkpoint folder just in case\n",
    "dirpath = 'checkpoint'\n",
    "if os.path.exists(dirpath) and os.path.isdir(dirpath):\n",
    "    shutil.rmtree(dirpath)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "010e988a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Deploy the sentiment analysis to kafka \n",
    "query = prediction \\\n",
    "    .writeStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", \"broker:29092\") \\\n",
    "    .option(\"topic\", \"sentiment_analysis\") \\\n",
    "    .option(\"checkpointLocation\", \"checkpoint/data\") \\\n",
    "    .start() \n",
    "# Important: delete existing checkpoints and not run memory query before"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a72b4b6c",
   "metadata": {},
   "source": [
    "# Step 4: Read/Query the prediction Stream"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "9a7d4a44",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read data from the topic \n",
    "df2 = spark \\\n",
    "  .readStream \\\n",
    "  .format(\"kafka\") \\\n",
    "  .option(\"kafka.bootstrap.servers\", \"broker:29092\") \\\n",
    "  .option(\"subscribe\", \"sentiment_analysis\") \\\n",
    "  .option(\"startingOffsets\",\"earliest\") \\\n",
    "  .option(\"includeHeaders\", \"true\") \\\n",
    "  .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "7c8e2156",
   "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",
      " |-- headers: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- key: string (nullable = true)\n",
      " |    |    |-- value: binary (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "baeb4bc5",
   "metadata": {},
   "source": [
    "Cleaning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "c82738b3",
   "metadata": {},
   "outputs": [],
   "source": [
    "string_stream_df = df2 \\\n",
    "    .withColumn(\"key\", df2[\"key\"].cast(StringType())) \\\n",
    "    .withColumn(\"value\", df2[\"value\"].cast(StringType()))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3b9f13df",
   "metadata": {},
   "source": [
    "Looking results in memory format"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "c9fba938",
   "metadata": {},
   "outputs": [],
   "source": [
    "see_results = string_stream_df \\\n",
    "    .writeStream \\\n",
    "    .outputMode(\"append\") \\\n",
    "    .format(\"memory\") \\\n",
    "    .queryName(\"predictions\") \\\n",
    "    .trigger(processingTime='2 seconds') \\\n",
    "    .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "c4ae779b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------------+---------+------+-----------------------+-------------+-------+\n",
      "|key |value                                                                                                                                                                                                                                                                                                                                                                        |topic             |partition|offset|timestamp              |timestampType|headers|\n",
      "+----+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------------+---------+------+-----------------------+-------------+-------+\n",
      "|null|{\"USER_NAME\":\"cleopatra 𖤐\",\"tweet_text\":\"need vaccine v2 so i can kiss cuties on the beach\",\"sentiment\":\"positive\"}                                                                                                                                                                                                                                                         |sentiment_analysis|0        |0     |2021-06-12 01:38:20.256|0            |null   |\n",
      "|null|{\"USER_NAME\":\"規制明け絶対民主法治国家維持\",\"tweet_text\":\"@jhengstler please don't mention the covid-19 daily per million cases of higher-vaccinatation-rate uk is beyond the those of much-lower-vaccination-rate india. https://t.co/iiEdDmCl2j\",\"sentiment\":\"negative\"}                                                                                                   |sentiment_analysis|0        |1     |2021-06-12 01:38:20.263|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Nina Gemsis\",\"tweet_text\":\"@DrP_MD @ashkaur Thankyou for saying this! Covid has completely opened my eyes to the realities going on around us. We have been lied to for years. I never saw it until now.\",\"sentiment\":\"negative\"}                                                                                                                              |sentiment_analysis|0        |2     |2021-06-12 01:38:20.264|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Ivermectin is a scam 🗯\",\"tweet_text\":\"@SkrimmyKat @BretWeinstein @HeatherEHeying Why don’t you ask your man Paul Marik, whose debunked vitamin C-based sepsis treatment was adapted for COVID with zero clinical basis.\",\"sentiment\":\"positive\"}                                                                                                              |sentiment_analysis|0        |3     |2021-06-12 01:38:20.266|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Hadenough\",\"tweet_text\":\"I also believe this will be how they will keep the propaganda going.\",\"sentiment\":\"positive\"}                                                                                                                                                                                                                                         |sentiment_analysis|0        |4     |2021-06-12 01:38:20.267|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Bill\",\"tweet_text\":\"@jsinijr @jack @Twitter @UltraRunningMag It was the April issue so I assume sometime in February. It was based on current vaccination rates but I have little respect for somebody that can't anticipate the process would scale up. Remember, the NY Times predicted 9+ years just to approve and make a vaccine.\",\"sentiment\":\"positive\"}|sentiment_analysis|0        |5     |2021-06-12 01:38:20.269|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Casey\",\"tweet_text\":\"@dd_chip @Transigence @maggieNYT I haven't been paying attention to this conversation, but your claim that all the people who died from COVID died because they didn't wear a mask or weren't as \\\"responsible\\\" as you thought they should be is incredibly offensive.\",\"sentiment\":\"positive\"}                                          |sentiment_analysis|0        |6     |2021-06-12 01:38:20.27 |0            |null   |\n",
      "|null|{\"USER_NAME\":\"Sgt Worrell #406\",\"tweet_text\":\"FDA says 60 million Johnson & Johnson vaccine doses from troubled Baltimore plant must be thrown out due to possible contamination https://t.co/q8hYG5qcVO\",\"sentiment\":\"negative\"}                                                                                                                                            |sentiment_analysis|0        |7     |2021-06-12 01:38:20.271|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Denise Mathurin\",\"tweet_text\":\"I’m attending Made to Save’s event, “For Our Future FL Weekly Zoom Phone Bank - Vaccine Month of Action Phone Bank” – sign up now to join me! #MadeToSave https://t.co/l8GncpcofQ\",\"sentiment\":\"positive\"}                                                                                                                      |sentiment_analysis|0        |8     |2021-06-12 01:38:20.272|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Sardar Khalsa\",\"tweet_text\":\"Where does the COVID Vaccine end up?  Not Where it Should Be https://t.co/WqFFYCQQGc\",\"sentiment\":\"negative\"}                                                                                                                                                                                                                     |sentiment_analysis|0        |9     |2021-06-12 01:38:20.273|0            |null   |\n",
      "|null|{\"USER_NAME\":\"John Maneri\",\"tweet_text\":\"@angrybklynmom COVID was a hoax. Have you not figured that out yet?  Most of us in TX got that back in May of 2020, hence your graduation experience. Go back home and vote the communists out of office in NY\",\"sentiment\":\"negative\"}                                                                                             |sentiment_analysis|0        |10    |2021-06-12 01:38:20.274|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Blue Eyes White Onion\",\"tweet_text\":\"The first birthday to celebrate with friends since the beginning of Covid. Happy birthday to my wonderful amazing spectacular girlfriend @BanditoTaylee https://t.co/rPo62T4A2j\",\"sentiment\":\"positive\"}                                                                                                                  |sentiment_analysis|0        |11    |2021-06-12 01:38:20.275|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Jimbo\",\"tweet_text\":\"@OregonGovBrown @flypdx @oregoncc Worst stats in the nation.  But sure, recall Kate brown.  👇\",\"sentiment\":\"positive\"}                                                                                                                                                                                                                   |sentiment_analysis|0        |12    |2021-06-12 01:38:20.277|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Albert Fong\",\"tweet_text\":\"Honeycomb's big! Yeah, yeah, yeah! A new vaccine could finally stop the decline in bee populations by breaking down the pesticide in the bee’s stomach before it reaches brain cells and causes damage https://t.co/Fh6nU1CFpg @adele_peters #bees #science #vaccine #nature #pesticides\",\"sentiment\":\"negative\"}                   |sentiment_analysis|0        |13    |2021-06-12 01:38:20.278|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Riley\",\"tweet_text\":\"HELL YES. KEEP THAT SD ENERGY ON THE SOUTH SIDE\",\"sentiment\":\"positive\"}                                                                                                                                                                                                                                                                  |sentiment_analysis|0        |14    |2021-06-12 01:38:20.279|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Andrea Allen\",\"tweet_text\":\"@shillman1 I’d take one of the cruises I’d have taken pre-COVID. A very small cruise, like under 100 people. A cruise along Alaska, for example. I’d never take one of those huge cruise ships (even before COVID).\",\"sentiment\":\"positive\"}                                                                                       |sentiment_analysis|0        |15    |2021-06-12 01:38:20.28 |0            |null   |\n",
      "|null|{\"USER_NAME\":\"Kim Climer\",\"tweet_text\":\"@Mike_BanksMD @drcraigwax @MonicaGandhi9 Look at this hypocrisy ⬇️ US patents Finnish developed Covid drug with ivermectin and hydroxychloroquine:     https://t.co/FtT94ccdsi\",\"sentiment\":\"positive\"}                                                                                                                              |sentiment_analysis|0        |16    |2021-06-12 01:38:20.281|0            |null   |\n",
      "|null|{\"USER_NAME\":\"The New Humanitarian\",\"tweet_text\":\"Legal liability could be a deal-breaker for a special supply of vaccines for the world’s trouble spots. https://t.co/ghPrVn7KHv\",\"sentiment\":\"positive\"}                                                                                                                                                                   |sentiment_analysis|0        |17    |2021-06-12 01:38:20.282|0            |null   |\n",
      "|null|{\"USER_NAME\":\"Zapsavvy\",\"tweet_text\":\"Flogging the dead imperial horse?\",\"sentiment\":\"negative\"}                                                                                                                                                                                                                                                                             |sentiment_analysis|0        |18    |2021-06-12 01:38:20.282|0            |null   |\n",
      "|null|{\"USER_NAME\":\"NWSFD\",\"tweet_text\":\"People & Profit - Nigerian finance minister on Covid-19: 'The vaccines are just not coming' - France24 / More stories at https://t.co/hkcunoILr5 https://t.co/A7mPpcd4kU\",\"sentiment\":\"negative\"}                                                                                                                                         |sentiment_analysis|0        |19    |2021-06-12 01:38:20.283|0            |null   |\n",
      "+----+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------------+---------+------+-----------------------+-------------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "None"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "clear_output(wait=True)\n",
    "display(spark.sql('SELECT * FROM predictions').show(20, False))\n",
    "time.sleep(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "956144f0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stop the spark session\n",
    "spark.stop()"
   ]
  }
 ],
 "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.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}