Sentiment_Analysis_Linear_Regression.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "sc=SparkContext()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark import SQLContext\n",
    "sqlContext=SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg import Vectors\n",
    "import json\n",
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "Load the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "mydir = ('file:/home/ubuntu/review.json')\n",
    "df1 = sqlContext.read.json(mydir)\n",
    "mydir2 = ('file:/home/ubuntu/business.json')\n",
    "df2 = sqlContext.read.json(mydir2)\n",
    "df = df1.join(df2,(df1.business_id==df2.business_id)).drop(df2.business_id).drop(df1.stars)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filter for chinese cuisine"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "asian=df.where(array_contains(df.categories,\"Chinese\"))\n",
    "asian.registerTempTable(\"asian\")\n",
    "asian_sample = sqlContext.sql(\"select business_id,text,stars from asian\")\n",
    "asian.registerTempTable(\"asian_sample\")\n",
    "asian_text=asian_sample.select('business_id','text')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filter the reviews for Service related aspects"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "words = [x.lower() for x in ['servic','time','price','lunch','friendli']]\n",
    "\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "def intersect(row):\n",
    "    # convert each word in lowercase\n",
    "    row = [x.lower() for x in row.split()]\n",
    "    return True if set(row).intersection(set(words)) else False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filtering sentences based on aspects and aggregate the sentences for each restaurant"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import nltk.data\n",
    "sent_detector = nltk.data.load('tokenizers/punkt/english.pickle')\n",
    "summarised=asian_text.rdd.map(lambda (business,text):(business,sent_detector.tokenize(text.strip())))\\\n",
    ".map(lambda (business,text):(business,\" \".join(filter(lambda sent: intersect(sent),text))))\\\n",
    ".reduceByKey(lambda x,y:x+y).toDF()\\\n",
    ".withColumnRenamed('_1','BusinessId').withColumnRenamed('_2','Review')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from textblob import TextBlob, Word, Blobber\n",
    "from textblob.classifiers import NaiveBayesClassifier\n",
    "from textblob.taggers import NLTKTagger\n",
    "asian_sample_rdd=summarised.rdd\n",
    "asian_sample_polarity = asian_sample_rdd.map(lambda x:(x[0],TextBlob(x[1]).sentiment.polarity))\n",
    "asian_pol1=asian_sample_polarity.toDF().withColumnRenamed('_1','business_id').withColumnRenamed('_2','pol1')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filter the reviews for Food related aspects"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "words = [x.lower() for x in ['noodl','dish','chicken','fri','rice','soup','sauc','beef','pork','tast']]\n",
    "from pyspark.sql.types import *\n",
    "def intersect(row):\n",
    "    # convert each word in lowercase\n",
    "    row = [x.lower() for x in row.split()]\n",
    "    return True if set(row).intersection(set(words)) else False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filtering sentences based on aspects and aggregate the sentences for each restaurant"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import nltk.data\n",
    "sent_detector = nltk.data.load('tokenizers/punkt/english.pickle')\n",
    "summarised=asian_text.rdd.map(lambda (business,text):(business,sent_detector.tokenize(text.strip())))\\\n",
    ".map(lambda (business,text):(business,\" \".join(filter(lambda sent: intersect(sent),text))))\\\n",
    ".reduceByKey(lambda x,y:x+y).toDF()\\\n",
    ".withColumnRenamed('_1','BusinessId').withColumnRenamed('_2','Review')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sentiment analysis using TextBlob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "asian_sample_rdd=summarised.rdd\n",
    "asian_sample_polarity = asian_sample_rdd.map(lambda x:(x[0],TextBlob(x[1]).sentiment.polarity))\n",
    "asian_pol2=asian_sample_polarity.toDF().withColumnRenamed('_1','business_id').withColumnRenamed('_2','pol2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Merge the topics for each restaurant"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "asian_rating = sqlContext.sql(\"select distinct business_id,stars from asian_sample\")\n",
    "temp = asian_rating.join(asian_pol1,(asian_rating.business_id==asian_pol1.business_id),'left').drop(asian_rating.business_id)\n",
    "asian_final=temp.join(asian_pol2,(temp.business_id==asian_pol2.business_id),'outer').drop(asian_pol2.business_id)\n",
    "asian_final=asian_final.withColumn('pol1',when(isnull(asian_final.pol1), 0).otherwise(asian_final.pol1))\n",
    "asian_final=asian_final.withColumn('pol2',when(isnull(asian_final.pol2), 0).otherwise(asian_final.pol2))asian_final=asian_final.filter(asian_final.business_id.isNotNull())\n",
    "asian_final.registerTempTable(\"asian_final1\")\n",
    "asian_final1=sqlContext.sql(\"select * from asian_final1 where not pol1 ==0 and not pol2==0\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Linear Regression Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.regression import LinearRegression\n",
    "training = asian_final1\n",
    "from pyspark.ml.feature import (VectorAssembler,VectorIndexer,\n",
    "                                OneHotEncoder,StringIndexer)\n",
    "numericCols = ['pol1','pol2']\n",
    "assembler = VectorAssembler(inputCols=numericCols, outputCol=\"features\")\n",
    "lr = LinearRegression(maxIter=10, featuresCol=\"features\",labelCol=\"stars\")\n",
    "training2=assembler.transform(training)\n",
    "# Fit the model\n",
    "lrModel = lr.fit(training2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Beta co-efficients"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "lrModel.coefficients"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "R-squared"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "lrModel.summary.r2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "p-values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "lrModel.summary.pValues"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.14"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}