pySpark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction \n",
    "\n",
    "- First we build a baseline model using basic Spark API. \n",
    "- Secon we fine tune the model using state of art technique using Spark NLP. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "_cell_guid": "b1076dfc-b9ad-4769-8c92-a6c4dae69d19",
    "_uuid": "8f2839f25d086af736a60e9eeb907d3b93b6e0e5"
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName('classification').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import lower, col, udf, lit, regexp_replace\n",
    "from pyspark.ml.feature import Tokenizer, StopWordsRemover, HashingTF, IDF, StringIndexer\n",
    "from pyspark.sql.types import IntegerType, StringType, StructType, StructField\n",
    "\n",
    "from pyspark.ml.classification import LogisticRegression, \\\n",
    "                RandomForestClassifier, NaiveBayes, LinearSVC, GBTClassifier\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "from pyspark.ml.pipeline import Pipeline\n",
    "\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv('Tweet_sentiment_clean.csv',\\\n",
    "                     header=True, inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "df1 = df.sample(0.1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+\n",
      "|sentiment|                text|\n",
      "+---------+--------------------+\n",
      "|        0|user dive mani ti...|\n",
      "|        0|spring break plai...|\n",
      "|        0|user day didnt ge...|\n",
      "+---------+--------------------+\n",
      "only showing top 3 rows\n",
      "\n",
      "Train data size:\t 161067\n"
     ]
    }
   ],
   "source": [
    "df1.show(3)\n",
    "print('Train data size:\\t', df1.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Train and test data schema\n",
      "root\n",
      " |-- sentiment: integer (nullable = true)\n",
      " |-- text: string (nullable = true)\n",
      "\n",
      "+---------+-----+\n",
      "|sentiment|count|\n",
      "+---------+-----+\n",
      "|        4|80630|\n",
      "|        0|80437|\n",
      "+---------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "print('Train and test data schema')\n",
    "df1.printSchema()\n",
    "\n",
    "df1.groupby('sentiment').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df1.where(df1['sentiment'].isNull()).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "index = StringIndexer(inputCol='sentiment', outputCol='target')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+\n",
      "|                text|target|\n",
      "+--------------------+------+\n",
      "|user dive mani ti...|   1.0|\n",
      "|spring break plai...|   1.0|\n",
      "|user day didnt ge...|   1.0|\n",
      "+--------------------+------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df1 = index.fit(df1).transform(df1).select('text', 'target')\n",
    "df1.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+\n",
      "|target|count|\n",
      "+------+-----+\n",
      "|   0.0|80630|\n",
      "|   1.0|80437|\n",
      "+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df1.groupby('target').count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Preprocessing "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocess(column):\n",
    "    \n",
    "    # Lowering the capital letters \n",
    "    column = lower(column)\n",
    "    \n",
    "    # Replacing user name \n",
    "    column = regexp_replace(column, r'@[^\\s]+', 'USER')\n",
    "    \n",
    "    # Replacing url \n",
    "    column = regexp_replace(column, r'https?://\\S+|www\\.\\S+', 'URL')\n",
    "    \n",
    "    # Replacing ayyyyy -> ayy \n",
    "    column = regexp_replace(column, r'(.)\\1\\1+', r'\\1\\1')\n",
    "    \n",
    "    # Replacing other than alphabets \n",
    "    column = regexp_replace(column, r'[^a-zA-Z\\d\\s]', '')\n",
    "    column = regexp_replace(column, r'\\d+', 'NUM')\n",
    "    \n",
    "    return column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+\n",
      "|                text|target|\n",
      "+--------------------+------+\n",
      "|user dive mani ti...|   1.0|\n",
      "|spring break plai...|   1.0|\n",
      "|user day didnt ge...|   1.0|\n",
      "+--------------------+------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df1 = df1.withColumn('text', preprocess(col('text')))\n",
    "df1.show(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Model building "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "train, test = df1.randomSplit([0.7, 0.3])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+\n",
      "|                text|target|\n",
      "+--------------------+------+\n",
      "|NUM good weekend sad|   1.0|\n",
      "|      NUM still done|   1.0|\n",
      "|NUM went fli toy ...|   1.0|\n",
      "+--------------------+------+\n",
      "only showing top 3 rows\n",
      "\n",
      "+--------------------+------+\n",
      "|                text|target|\n",
      "+--------------------+------+\n",
      "|NUM last night ka...|   1.0|\n",
      "|NUMhh friday fune...|   1.0|\n",
      "|NUMrgh hate cop n...|   1.0|\n",
      "+--------------------+------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "(48129, 112938)"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train.show(3)\n",
    "test.show(3)\n",
    "test.count(), train.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total number of stopwords: \t 181\n",
      "Example stopwords: \t\t ['i', 'me', 'my', 'myself', 'we']\n"
     ]
    }
   ],
   "source": [
    "print('Total number of stopwords: \\t', len(StopWordsRemover().getStopWords()))\n",
    "print('Example stopwords: \\t\\t', StopWordsRemover().getStopWords()[:5])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "evaluator = MulticlassClassificationEvaluator(labelCol='target', metricName='accuracy')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "tokenizer = Tokenizer(inputCol='text', outputCol='text_token')\n",
    "remover = StopWordsRemover(inputCol='text_token', outputCol='text_trim')\n",
    "\n",
    "hashingTF = HashingTF(inputCol='text_trim', outputCol='raw_feat', numFeatures=10000)\n",
    "idf = IDF(inputCol='raw_feat', outputCol='features')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "def pipeline(model):\n",
    "    pipeline_model = Pipeline(stages=[tokenizer, remover, hashingTF, idf, model])\n",
    "    return pipeline_model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 132,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.7393231180885089\n"
     ]
    }
   ],
   "source": [
    "ridge = LogisticRegression(labelCol='target', \n",
    "                        maxIter=20, \n",
    "                        regParam=0.7, \n",
    "                        elasticNetParam=0)\n",
    "\n",
    "pipeline_ridge = pipeline(ridge)\n",
    "\n",
    "model = pipeline_ridge.fit(train)\n",
    "pred = model.transform(test)\n",
    "score = evaluator.evaluate(pred)\n",
    "print(score)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.7513849112611577\n"
     ]
    }
   ],
   "source": [
    "lasso = LogisticRegression(labelCol='target', \n",
    "                        maxIter=20, \n",
    "                        regParam=0.002, \n",
    "                        elasticNetParam=1)\n",
    "\n",
    "pipeline_lasso  = pipeline(lasso)\n",
    "\n",
    "model = pipeline_lasso.fit(train)\n",
    "pred = model.transform(test)\n",
    "score = evaluator.evaluate(pred)\n",
    "print(score)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 141,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.7439220686915986\n"
     ]
    }
   ],
   "source": [
    "nb = NaiveBayes(labelCol='target', smoothing=400)\n",
    "\n",
    "pipeline_nb = pipeline(nb)\n",
    "\n",
    "model = pipeline_nb.fit(train)\n",
    "pred = model.transform(test)\n",
    "score = evaluator.evaluate(pred)\n",
    "print(score)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.6990172245423757\n"
     ]
    }
   ],
   "source": [
    "rf = RandomForestClassifier(labelCol='target', \n",
    "                            maxDepth=30,\n",
    "                            maxBins=5,\n",
    "                            numTrees=10)\n",
    "\n",
    "pipeline_rf = pipeline(rf)\n",
    "\n",
    "model = pipeline_rf.fit(train)\n",
    "pred = model.transform(test)\n",
    "score = evaluator.evaluate(pred)\n",
    "print(score)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.6752269941199692\n"
     ]
    }
   ],
   "source": [
    "gb = GBTClassifier(labelCol='target')\n",
    "\n",
    "pipeline_gb = pipeline(gb)\n",
    "\n",
    "model = pipeline_gb.fit(train)\n",
    "pred = model.transform(test)\n",
    "score = evaluator.evaluate(pred)\n",
    "print(score)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Grid search "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "nb = NaiveBayes(labelCol='target')\n",
    "pipeline_nb = pipeline(nb)\n",
    "\n",
    "paramGrid = ParamGridBuilder() \\\n",
    "    .addGrid(hashingTF.numFeatures, [10000, 100000]) \\\n",
    "    .addGrid(nb.smoothing, [200, 400, 600]) \\\n",
    "    .build()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Naive Bayes                    0.7753605642372429\n"
     ]
    }
   ],
   "source": [
    "model_names = {'Naive Bayes' : pipeline_nb}\n",
    "\n",
    "for name in model_names:\n",
    "    crossval = CrossValidator(estimator=model_names[name],\n",
    "                              estimatorParamMaps=paramGrid,\n",
    "                              evaluator=evaluator,\n",
    "                              numFolds=5)\n",
    "    \n",
    "    model = crossval.fit(df1)\n",
    "    pred = model.transform(df1)\n",
    "    score = evaluator.evaluate(pred)\n",
    "    print(name.ljust(30), score)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Future direction \n",
    "\n",
    "- Emoji and emoticons \n",
    "- Stacking in pipeline "
   ]
  },
  {
   "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}