twitter_hate_speech.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "from pyspark.sql.session import SparkSession\n",
    "from pyspark.streaming import StreamingContext\n",
    "import pyspark.sql.types as tp\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import StringIndexer, OneHotEncoderEstimator, VectorAssembler\n",
    "from pyspark.ml.feature import StopWordsRemover, Word2Vec, RegexTokenizer\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.sql import Row\n",
    "import sys\n",
    "from pyspark.streaming.kafka import KafkaUtils\n",
    "import json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create session\n",
    "\n",
    "spark = SparkSession \\\n",
    " .builder \\\n",
    " .appName(\"tweeter_hate_speeach\") \\\n",
    " .master(\"local\") \\\n",
    " .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define the schema\n",
    "# spark availabe types: see https://spark.apache.org/docs/1.6.0/api/java/org/apache/spark/sql/types/package-summary.html\n",
    "my_schema = tp.StructType([\n",
    "  tp.StructField(name= 'id',          dataType= tp.IntegerType(),  nullable= True),\n",
    "  tp.StructField(name= 'label',       dataType= tp.IntegerType(),  nullable= True),\n",
    "  tp.StructField(name= 'tweet',       dataType= tp.StringType(),   nullable= True)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# read the dataset  \n",
    "my_data = spark.read.csv('data/twitter_sentiments.csv',\n",
    "                         schema=my_schema,\n",
    "                         header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+--------------------+\n",
      "| id|label|               tweet|\n",
      "+---+-----+--------------------+\n",
      "|  1|    0| @user when a fat...|\n",
      "|  2|    0|@user @user thank...|\n",
      "|  3|    0|  bihday your maj...|\n",
      "|  4|    0|#model   i love u...|\n",
      "|  5|    0| factsguide: soci...|\n",
      "+---+-----+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# view the data\n",
    "my_data.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: integer (nullable = true)\n",
      " |-- label: integer (nullable = true)\n",
      " |-- tweet: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# print the schema of the file\n",
    "my_data.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "31962"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# count data\n",
    "my_data.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# create pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## pipeline steps\n",
    "**load data** -> **tokenize** -> **remove stop words** -> **word vector** -> **logistic regression** -> **predicted labels**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define stage 1: tokenize the tweet text    \n",
    "stage_1 = RegexTokenizer(inputCol= 'tweet' , outputCol= 'tokens', pattern= '\\\\W')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define stage 2: remove the stop words\n",
    "stage_2 = StopWordsRemover(inputCol= 'tokens', outputCol= 'filtered_words')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define stage 3: create a word vector of the size 100\n",
    "stage_3 = Word2Vec(inputCol= 'filtered_words', outputCol= 'vector', vectorSize= 100)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define stage 4: Logistic Regression Model\n",
    "model = LogisticRegression(featuresCol= 'vector', labelCol= 'label')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# setup the pipeline\n",
    "pipeline = Pipeline(stages= [stage_1, stage_2, stage_3, model])\n",
    "\n",
    "# fit the pipeline model with the training data\n",
    "pipelineFit = pipeline.fit(my_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# streaming predictions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define a function to compute sentiments of the received tweets\n",
    "def get_prediction(tweet_text):\n",
    "\ttry:\n",
    "    # filter the tweets whose length is greater than 0\n",
    "\t\ttweet_text = tweet_text.filter(lambda x: len(x) > 0)\n",
    "    # create a dataframe with column name 'tweet' and each row will contain the tweet\n",
    "\t\trowRdd = tweet_text.map(lambda w: Row(tweet=w))\n",
    "    # create a spark dataframe\n",
    "\t\twordsDataFrame = spark.createDataFrame(rowRdd)\n",
    "    # transform the data using the pipeline and get the predicted sentiment\n",
    "\t\tpipelineFit.transform(wordsDataFrame).select('tweet','prediction').show()\n",
    "\texcept : \n",
    "\t\tprint('No data')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# initialize the streaming context \n",
    "ssc = StreamingContext(SparkContext.getOrCreate(), batchDuration= 3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a DStream that will connect to hostname:port, like localhost:9991\n",
    "#lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))\n",
    "lines = ssc.socketTextStream('localhost', 9999)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\"kafkaStream = KafkaUtils.createStream(ssc, \\\n",
    "#     'localhost:2181', 'group_01', {'twitter_hater':1})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# split the tweet text by a keyword 'TWEET_APP' so that we can identify which set of words is from a single tweet\n",
    "words = lines.flatMap(lambda line : line.split('TWEET_APP'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get the predicted sentiments for the tweets received\n",
    "words.foreachRDD(get_prediction)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n"
     ]
    }
   ],
   "source": [
    "# Start the computation\n",
    "ssc.start()   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n",
      "No data\n"
     ]
    }
   ],
   "source": [
    "# Wait for the computation to terminate\n",
    "ssc.awaitTermination()  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-9-7084fe453b13>, line 1)",
     "output_type": "error",
     "traceback": [
      "\u001b[0;36m  File \u001b[0;32m\"<ipython-input-9-7084fe453b13>\"\u001b[0;36m, line \u001b[0;32m1\u001b[0m\n\u001b[0;31m    https://www.howtoforge.com/how-to-setup-apache-zookeeper-cluster-on-ubuntu-1804/\u001b[0m\n\u001b[0m           ^\u001b[0m\n\u001b[0;31mSyntaxError\u001b[0m\u001b[0;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "https://www.howtoforge.com/how-to-setup-apache-zookeeper-cluster-on-ubuntu-1804/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "https://www.tutorialspoint.com/apache_kafka/apache_kafka_basic_operations.htm"
   ]
  }
 ],
 "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.6.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}