TwitterSentimentAnalysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg import Vectors\n",
    "from pyspark.ml.feature import StopWordsRemover,Tokenizer\n",
    "from pyspark.sql.functions import lower\n",
    "from pyspark.sql.types import Row\n",
    "from pyspark.sql.functions import monotonically_increasing_id\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark import SparkContext\n",
    "\n",
    "import re\n",
    "from collections import Counter\n",
    "from nltk.stem import PorterStemmer\n",
    "from math import log10,sqrt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark = SparkSession\\\n",
    "        .builder\\\n",
    "        .appName(\"NaiveBayesExample\")\\\n",
    "        .getOrCreate()\n",
    "\n",
    "data = spark.read.csv(\"Sentiment Analysis Dataset.csv\",header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "stemmer = PorterStemmer()\n",
    "def parse(sentence):\n",
    "    blob = [re.sub('[^a-zA-Z]+', '', s) for s in sentence]\n",
    "    blob = [stemmer.stem(i) for i in blob]\n",
    "    blob = [i.lower() for i in blob if len(i)>2 and len(i)<20]\n",
    "    return blob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Converting senteces to words\n",
    "tokenizer = Tokenizer(inputCol=\"SentimentText\", outputCol=\"words\")\n",
    "token_df=tokenizer.transform(data).drop(\"SentimentText\")\n",
    "\n",
    "#Removing stop words \n",
    "remover=StopWordsRemover(inputCol=\"words\", outputCol=\"filtered_tweet\")\n",
    "data_stop_words=remover.transform(token_df).drop(\"words\")\\\n",
    "                       .select(\"*\").withColumn(\"id\", monotonically_increasing_id())\n",
    "\n",
    "\n",
    "data_stop_words_rdd=data_stop_words.rdd       #transforming to RDD\n",
    "\n",
    "#Spliting into train and test data\n",
    "train,test=data_stop_words_rdd.randomSplit([0.7,0.3])\n",
    "\n",
    "train_data_tweets=train.map(lambda row: (\"doc\"+str(row.id),parse(row.filtered_tweet)))     # RDD -> (doc_id,tweet)      \n",
    "train_labels=train.map(lambda row:(\"doc\"+str(row.id),row.Sentiment)) # RDD -> (doc_id, sentiment)\n",
    "\n",
    "test_data_tweets=test.map(lambda row: (\"testDoc\"+str(row.id),parse(row.filtered_tweet)))    # RDD -> (doc_id, tweet)\n",
    "test_labels=test.map(lambda row:(\"testDoc\"+str(row.id),row.Sentiment))   # RDD -> (doc_id, sentiment)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('doc0', ['sad', 'apl', 'friend']),\n",
       " ('doc1', ['miss', 'new', 'moon', 'trailer']),\n",
       " ('doc2', ['omg', 'alreadi']),\n",
       " ('doc6', ['chillin']),\n",
       " ('doc9', ['hmmmm', 'wonder', 'number'])]"
      ]
     },
     "execution_count": 80,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_data_tweets.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def term_freq(row):\n",
    "    doc_id=row[0]\n",
    "    doc=row[1]\n",
    "    a=Counter(doc).items()\n",
    "    return [(k[0],(1+log10(k[1]),doc_id)) for k in a]           # TF= 1+log(WF); RDD -> (term,(TF,doc_id))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "train_tf=train_data_tweets.flatMap(term_freq)  # RDD -> (term,(TF,doc_id))\n",
    "test_tf=test_data_tweets.flatMap(term_freq)    # RDD -> (term,(TF,doc_id))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('sad', (1.0, 'doc0')),\n",
       " ('apl', (1.0, 'doc0')),\n",
       " ('friend', (1.0, 'doc0')),\n",
       " ('miss', (1.0, 'doc1')),\n",
       " ('trailer', (1.0, 'doc1')),\n",
       " ('new', (1.0, 'doc1')),\n",
       " ('moon', (1.0, 'doc1')),\n",
       " ('alreadi', (1.0, 'doc2'))]"
      ]
     },
     "execution_count": 83,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_tf.take(8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def tfidfCalc(row):\n",
    "    tf=row[1][1][0]\n",
    "    idf=log10(1+(num_docs/row[1][0]))\n",
    "    doc_id=row[1][1][1]\n",
    "    return (doc_id,(row[0],tf*idf))                    #Mulitplying TF and IDF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('noelpresid', 1), ('nite', 11), ('mad', 23), ('iwi', 1)]"
      ]
     },
     "execution_count": 85,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_tf.map(lambda row:(row[0],1)).reduceByKey(lambda a,b:a+b).take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "train_doc_count=train_tf.map(lambda row:(row[0],1)).reduceByKey(lambda x,y: x+y)   # RDD -> (term, Number of documents containing the term t)\n",
    "# print(train_doc_count.take(5))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "num_docs=train.count()\n",
    "train_tfidf=train_doc_count.join(train_tf)\\\n",
    "                           .map(tfidfCalc)       #RDD -> (term,(doc_count,())) => #RDD -> (doc_id,(term,tfidf))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('doc2227', ('noelpresid', 3.8477576883923312)),\n",
       " ('doc430', ('nite', 2.8069811986863766)),\n",
       " ('doc471', ('nite', 2.8069811986863766)),\n",
       " ('doc1472', ('nite', 2.8069811986863766)),\n",
       " ('doc1525', ('nite', 2.8069811986863766)),\n",
       " ('doc1661', ('nite', 2.8069811986863766)),\n",
       " ('doc1741', ('nite', 2.8069811986863766)),\n",
       " ('doc2267', ('nite', 2.8069811986863766))]"
      ]
     },
     "execution_count": 88,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_tfidf.take(8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "denom_square=train_tfidf.map(lambda row:(row[0],row[1][1])).reduceByKey(lambda a,b:a*a+b*b) #RDD -> (doc_id,denom_square)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('doc7049', (1.2044510733559554e+79, ('episod', 2.8069811986863766))),\n",
       " ('doc7049', (1.2044510733559554e+79, ('fave', 2.8940082052377174))),\n",
       " ('doc7049', (1.2044510733559554e+79, ('know', 1.6081171952607451))),\n",
       " ('doc7049', (1.2044510733559554e+79, ('let', 1.9495119823462912)))]"
      ]
     },
     "execution_count": 90,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "denom_square.join(train_tfidf).take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def norm_tfidf(row):\n",
    "    tf_idf=row[1][1][1]/sqrt(row[1][0])\n",
    "    return (row[0],(row[1][1][0],tf_idf))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('doc7049', ('episod', 8.088070389699966e-40)),\n",
       " ('doc7049', ('fave', 8.338831084186102e-40)),\n",
       " ('doc7049', ('know', 4.633648802579319e-40)),\n",
       " ('doc7049', ('let', 5.61734797018213e-40))]"
      ]
     },
     "execution_count": 92,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_tfidf_norm=denom_square.join(train_tfidf).map(norm_tfidf)# RDD -> (doc_id,())\n",
    "train_tfidf_norm.take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "train_tfidf_norm_labels=train_tfidf_norm.join(train_labels)     \n",
    "#RDD -> (doc_id,((term,tfidf),label))\n",
    "#train_tfidf_norm_labels.collect()                                   #RDD -> (doc_id,((term,tfidf),label))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Naive Bayes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "alpha=train_tfidf.count()     #Vocabulary\n",
    "alpha"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "labels_count=train.map(lambda row:(row.Sentiment,1))\\\n",
    "                                    .reduceByKey(lambda a,b:a+b).collect()\n",
    "labels_count=dict([(lab,c/train.count()) for lab,c in labels_count])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "B=dict(train_tfidf_norm_labels.map(lambda row:(row[1][1],row[1][0][1]))\\\n",
    "                              .reduceByKey(lambda a,b:a+b).collect())            #Number of words in each class\n",
    "print(B)\n",
    "B_alpha={k: v +alpha for k, v in B.items()}\n",
    "B_alpha"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# (doc_id,((term,tfidf),label))=> ((term,label),tfidf) => (term,(label,prob=(1+tfidf)/(alpha+label_count)))\n",
    "prior=train_tfidf_norm_labels.map(lambda row:((row[1][0][0],row[1][1]),row[1][0][1]))\\\n",
    "                     .reduceByKey(lambda a,b:a+b)\\\n",
    "                     .map(lambda row:((row[0][0],row[0][1]),(1+row[1])/B_alpha[row[0][1]]))\n",
    "print(\"prior\",prior.take(5))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# format for join of tes_tf and prior is [(term,((term_freq,doc_id),(label,prob)))]\n",
    "test_prior_negative=test_tf.map(lambda row: ((row[0],'0'),(row[1][0],row[1][1]))).leftOuterJoin(prior)\n",
    "test_prior_positive=test_tf.map(lambda row: ((row[0],'1'),(row[1][0],row[1][1]))).leftOuterJoin(prior)\n",
    "# print(\"test_prior_negative\",test_prior_negative.collect())\n",
    "# print(\"test_prior_positive\",test_prior_positive.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def checkForNone(lab,val):\n",
    "    if(val==None):\n",
    "        return 1/B_alpha[lab]\n",
    "    else:\n",
    "        return val\n",
    "    \n",
    "#Multiplying probabilities of terms in each doc    \n",
    "test_negative_docs=test_prior_negative.map(lambda row: (row[1][0][1],checkForNone(row[0][1],row[1][1]))).reduceByKey(lambda a,b:a*b)\n",
    "test_positive_docs=test_prior_positive.map(lambda row: (row[1][0][1],checkForNone(row[0][1],row[1][1]))).reduceByKey(lambda a,b:a*b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# print(\"test_negative_docs\",test_negative_docs.collect())\n",
    "# print(\"test_positive_docs\",test_positive_docs.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "predicted_values=test_negative_docs.join(test_positive_docs).map(lambda row: (row[0],'0' if row[1][0]>row[1][1] else '1'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "accuracy=dict(predicted_values.join(test_labels).map(lambda row: (True,1) if row[1][0]==row[1][1] else (False,1)).reduceByKey(lambda a,b:a+b).collect())\n",
    "accuracy=100*accuracy[True]/sum(accuracy.values())\n",
    "print(str(data.count())+\"=>\"+str(accuracy))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# KNN\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def test_tfidfCalc(tf,df):\n",
    "    idf=log10(1+(num_docs+1/df+1))\n",
    "    return tf*idf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "test_tfidf=test_tf.join(train_doc_count).map(lambda row:(row[1][0][1],(row[0],test_tfidfCalc(row[1][0][0],row[1][1]))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# test_tfidf.take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "test_denom_square=test_tfidf.map(lambda row:(row[0],row[1][1])).reduceByKey(lambda a,b:a*a+b*b) #RDD -> (doc_id,denom_square)\n",
    "# test_denom_square.take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def test_norm_tfidf(row):\n",
    "    tf_idf=row[1][1][1]/sqrt(row[1][0])\n",
    "    return (row[1][1][0],(row[0],tf_idf))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "test_tfidf_norm=test_denom_square.join(test_tfidf).map(test_norm_tfidf)# RDD -> (doc_id,())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "scrolled": true
   },
   "source": [
    "train_tfidf_norm.take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "test_train=test_tfidf_norm.join(train_tfidf_norm.map(lambda row:(row[1][0],(row[0],row[1][1]))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# test_train.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "test_scores=test_train.map(lambda row: ((row[1][0][0],row[1][1][0]),row[1][0][1]*row[1][1][1]))\\\n",
    "                           .reduceByKey(lambda a,b:a+b)\n",
    "# test_scores.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "test_sorted=test_scores.map(lambda row:(row[0][0],(row[0][1],row[1])))\n",
    "# test_sorted.take(9)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# test_sorted.filter(lambda row:row[0]=='testDoc10').take(9)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "predicted=[]\n",
    "out=[]\n",
    "for i in set(test_sorted.keys().collect()):\n",
    "    top_9_docs=dict(test_sorted.filter(lambda row:row[0]==i).map(lambda row:(row[1][0],row[1][1])).sortBy(lambda row:row[0],ascending=False).take(3))\n",
    "    top_9_docs_labels=train_labels.filter(lambda row:row[0] in top_9_docs.keys())\n",
    "    labels_count=dict(top_9_docs_labels.map(lambda row:(row[1],1)).reduceByKey(lambda a,b:a+b).collect())\n",
    "    test_output=max(labels_count,key=labels_count.get)\n",
    "    out.append(test_output)\n",
    "    predicted.append((i,test_output))\n",
    "predictedRDD=sc.parallelize(predicted)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "accuracy=100*(dict(predictedRDD.join(test_labels).map(lambda row: (True,1) if row[1][0]==row[1][1] else (False,1))\\\n",
    "                                   .reduceByKey(lambda a,b:a+b).collect())[True])/test_labels.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "accuracy"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [conda root]",
   "language": "python",
   "name": "conda-root-py"
  },
  "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}