pyspark_logistic_reg_reddit.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Twitter sentiment analysis and prediction using pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython import display\n",
    "import math\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark import SparkContext\n",
    "\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://www.learndatasci.com/tutorials/sentiment-analysis-reddit-headlines-pythons-nltk/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc =SparkContext()\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "customSchema = StructType([\n",
    "    StructField(\"clean_text\", StringType()), \n",
    "    StructField(\"category\", StringType())])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "#modi_data.csv file contains 10000 tweets with seach query modi\n",
    "filename = 'redt_dataset.csv'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = sqlContext.read.format(\"csv\").option(\"header\", \"true\").schema(customSchema).load(filename)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------+\n",
      "|          clean_text|category|\n",
      "+--------------------+--------+\n",
      "|though most the u...|       1|\n",
      "| parents live nea...|       1|\n",
      "|jets flying reall...|       1|\n",
      "| feel for the bra...|       1|\n",
      "| switched republi...|      -1|\n",
      "+--------------------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data = df.na.drop(how='any')\n",
    "data.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- clean_text: string (nullable = true)\n",
      " |-- category: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Preprocessing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-----+\n",
      "|category|count|\n",
      "+--------+-----+\n",
      "|       1|15749|\n",
      "|       0|12895|\n",
      "|      -1| 8244|\n",
      "+--------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import col\n",
    "\n",
    "data.groupBy(\"category\").count().orderBy(col(\"count\").desc()).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model Pipeline\n",
    "Spark Machine Learning Pipelines API is similar to Scikit-Learn. Our pipeline includes three steps:\n",
    "\n",
    "regexTokenizer: Tokenization (with Regular Expression)\n",
    "\n",
    "stopwordsRemover: Remove Stop Words\n",
    "\n",
    "countVectors: Count vectors (“document-term vectors”)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import RegexTokenizer, StopWordsRemover, CountVectorizer\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "# regular expression tokenizer\n",
    "regexTokenizer = RegexTokenizer(inputCol=\"clean_text\", outputCol=\"words\", pattern=\"\\\\W\")\n",
    "\n",
    "# stop words\n",
    "add_stopwords = [\"http\",\"https\",\"amp\",\"rt\",\"t\",\"c\",\"the\"] \n",
    "\n",
    "stopwordsRemover = StopWordsRemover(inputCol=\"words\", outputCol=\"filtered\").setStopWords(add_stopwords)\n",
    "\n",
    "# bag of words count\n",
    "countVectors = CountVectorizer(inputCol=\"filtered\", outputCol=\"features\", vocabSize=15000, minDF=5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------+--------------------+--------------------+--------------------+-----+\n",
      "|          clean_text|category|               words|            filtered|            features|label|\n",
      "+--------------------+--------+--------------------+--------------------+--------------------+-----+\n",
      "|though most the u...|       1|[though, most, th...|[though, most, us...|(12890,[0,1,2,3,4...|  0.0|\n",
      "| parents live nea...|       1|[parents, live, n...|[parents, live, n...|(12890,[6,37,88,9...|  0.0|\n",
      "|jets flying reall...|       1|[jets, flying, re...|[jets, flying, re...|(12890,[36,68,73,...|  0.0|\n",
      "| feel for the bra...|       1|[feel, for, the, ...|[feel, for, brave...|(12890,[3,14,19,3...|  0.0|\n",
      "| switched republi...|      -1|[switched, republ...|[switched, republ...|(12890,[2,3,4,6,1...|  2.0|\n",
      "+--------------------+--------+--------------------+--------------------+--------------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler\n",
    "label_stringIdx = StringIndexer(inputCol = \"category\", outputCol = \"label\")\n",
    "\n",
    "pipeline = Pipeline(stages=[regexTokenizer, stopwordsRemover, countVectors, label_stringIdx])\n",
    "\n",
    "# Fit the pipeline to training documents.\n",
    "pipelineFit = pipeline.fit(data)\n",
    "dataset = pipelineFit.transform(data)\n",
    "dataset.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Partition Training & Test sets¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Training Dataset Count: 25905\n",
      "Test Dataset Count: 10983\n"
     ]
    }
   ],
   "source": [
    "# set seed for reproducibility\n",
    "(trainingData, testData) = dataset.randomSplit([0.7, 0.3], seed = 100)\n",
    "print(\"Training Dataset Count: \" + str(trainingData.count()))\n",
    "print(\"Test Dataset Count: \" + str(testData.count()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model Training and Evaluation\n",
    "Logistic Regression using Count Vector Features \n",
    "\n",
    "Our model will make predictions and score on the test set; we then look at the top 10 predictions from the highest probability."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------+--------+------------------------------+-----+----------+\n",
      "|                    clean_text|category|                   probability|label|prediction|\n",
      "+------------------------------+--------+------------------------------+-----+----------+\n",
      "| modi’ government last four...|       1|[0.9999998432903444,1.43071...|  0.0|       0.0|\n",
      "|she right she right she rig...|       1|[0.9999997579988142,3.62693...|  0.0|       0.0|\n",
      "|great job mellowde writing ...|       1|[0.9999997416777889,3.39726...|  0.0|       0.0|\n",
      "| very interested history an...|       1|[0.9999986290163623,6.14373...|  0.0|       0.0|\n",
      "|just tried organize the pol...|      -1|[0.9999981355577869,6.46211...|  2.0|       0.0|\n",
      "| chennai super kings ipl 20...|       1|[0.9999965535336731,3.68668...|  0.0|       0.0|\n",
      "|this the first time was gre...|       1|[0.9999955444739944,3.64267...|  0.0|       0.0|\n",
      "|there couple reasons see fo...|       1|[0.9999934766327285,7.64619...|  0.0|       0.0|\n",
      "|’ gunning for rcb but think...|       1|[0.9999898190034273,2.21671...|  0.0|       0.0|\n",
      "| another child the while ad...|       1|[0.9999895898760462,2.18227...|  0.0|       0.0|\n",
      "+------------------------------+--------+------------------------------+-----+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "lr = LogisticRegression(maxIter=20, regParam=0.3, elasticNetParam=0)\n",
    "lrModel = lr.fit(trainingData)\n",
    "\n",
    "predictions = lrModel.transform(testData)\n",
    "\n",
    "predictions.filter(predictions['prediction'] == 0).select(\"clean_text\",\"category\",\"probability\",\"label\",\"prediction\")\\\n",
    ".orderBy(\"probability\", ascending=False).show(n = 10, truncate = 30)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.697852576519381"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "evaluator = MulticlassClassificationEvaluator(predictionCol=\"prediction\")\n",
    "evaluator.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Logistic Regression using TF-IDF Features¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------+--------+------------------------------+-----+----------+\n",
      "|                    clean_text|category|                   probability|label|prediction|\n",
      "+------------------------------+--------+------------------------------+-----+----------+\n",
      "|just tried organize the pol...|      -1|[0.9999999998459883,5.31840...|  2.0|       0.0|\n",
      "|she right she right she rig...|       1|[0.9999997245487119,2.28599...|  0.0|       0.0|\n",
      "| take rather separated from...|       1|[0.9999979078617137,2.33964...|  0.0|       0.0|\n",
      "|translations armor names hu...|      -1|[0.9999965982307848,3.39439...|  2.0|       0.0|\n",
      "|great job mellowde writing ...|       1|[0.999996386371032,2.457027...|  0.0|       0.0|\n",
      "| chennai super kings ipl 20...|       1|[0.9999925927459271,9.31544...|  0.0|       0.0|\n",
      "|typical elitist indian mind...|       1|[0.9999788261906685,1.33554...|  0.0|       0.0|\n",
      "| which team are you gunning...|       1|[0.9999775608662237,3.86977...|  0.0|       0.0|\n",
      "|great post this really rele...|       1|[0.9999757329179939,1.17003...|  0.0|       0.0|\n",
      "| which team are you gunning...|       1|[0.9999754315797097,4.34172...|  0.0|       0.0|\n",
      "+------------------------------+--------+------------------------------+-----+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.feature import HashingTF, IDF\n",
    "\n",
    "hashingTF = HashingTF(inputCol=\"filtered\", outputCol=\"rawFeatures\", numFeatures=10000)\n",
    "idf = IDF(inputCol=\"rawFeatures\", outputCol=\"features\", minDocFreq=5) #minDocFreq: remove sparse terms\n",
    "pipeline = Pipeline(stages=[regexTokenizer, stopwordsRemover, hashingTF, idf, label_stringIdx])\n",
    "\n",
    "pipelineFit = pipeline.fit(data)\n",
    "dataset = pipelineFit.transform(data)\n",
    "\n",
    "(trainingData, testData) = dataset.randomSplit([0.7, 0.3], seed = 100)\n",
    "lr = LogisticRegression(maxIter=20, regParam=0.3, elasticNetParam=0)\n",
    "lrModel = lr.fit(trainingData)\n",
    "\n",
    "predictions = lrModel.transform(testData)\n",
    "\n",
    "predictions.filter(predictions['prediction'] == 0) \\\n",
    "    .select(\"clean_text\",\"category\",\"probability\",\"label\",\"prediction\") \\\n",
    "    .orderBy(\"probability\", ascending=False) \\\n",
    "    .show(n = 10, truncate = 30)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.6736147017995611"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "evaluator = MulticlassClassificationEvaluator(predictionCol=\"prediction\")\n",
    "evaluator.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cross-Validation\n",
    "Let’s now try cross-validation to tune our hyper parameters, and we will only tune the count vectors Logistic Regression."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline = Pipeline(stages=[regexTokenizer, stopwordsRemover, countVectors, label_stringIdx])\n",
    "\n",
    "pipelineFit = pipeline.fit(data)\n",
    "dataset = pipelineFit.transform(data)\n",
    "(trainingData, testData) = dataset.randomSplit([0.7, 0.3], seed = 100)\n",
    "\n",
    "lr = LogisticRegression(maxIter=20, regParam=0.3, elasticNetParam=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.7449757330752173"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "\n",
    "# Create ParamGrid for Cross Validation\n",
    "paramGrid = (ParamGridBuilder()\n",
    "             .addGrid(lr.regParam, [0.1, 0.3, 0.5]) # regularization parameter\n",
    "             .addGrid(lr.elasticNetParam, [0.0, 0.1, 0.2]) # Elastic Net Parameter (Ridge = 0)\n",
    "#            .addGrid(model.maxIter, [10, 20, 50]) #Number of iterations\n",
    "#            .addGrid(idf.numFeatures, [10, 100, 1000]) # Number of features\n",
    "             .build())\n",
    "\n",
    "# Create 5-fold CrossValidator\n",
    "cv = CrossValidator(estimator=lr, \\\n",
    "                    estimatorParamMaps=paramGrid, \\\n",
    "                    evaluator=evaluator, \\\n",
    "                    numFolds=5)\n",
    "\n",
    "cvModel = cv.fit(trainingData)\n",
    "\n",
    "predictions = cvModel.transform(testData)\n",
    "# Evaluate best model\n",
    "evaluator = MulticlassClassificationEvaluator(predictionCol=\"prediction\")\n",
    "evaluator.evaluate(predictions)\n",
    "#print(\"Test Area Under ROC: \" + str(evaluator.evaluate(predictions, {evaluator.metricName: \"areaUnderROC\"})))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}