Multi-class Text Classification Problem with PySpark and MLlib.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Multi-class Text Classification Problem with PySpark and MLlib"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Background\n",
    "Apache Spark is quickly gaining steam both in the headlines and real-world adoption, mainly because of its ability to process streaming data. With so much data being processed on a daily basis, it has become essential for us to be able to stream and analyze it in real time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We use Spark Machine Learning Library (Spark MLlib) to solve this multi-class text classification problem"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Load Data \n",
    "The dataset is from Kaggle [Link](https://www.kaggle.com/c/sf-crime/data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The problem is to classify 'Crime Description' into 33 categories"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "#we use the findspark library to locate spark on our local machine\n",
    "import findspark\n",
    "findspark.init('C:/Users/bokhy/spark/spark-2.4.6-bin-hadoop2.7')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark # only run after findspark.init()\n",
    "from pyspark.sql import SparkSession, SQLContext\n",
    "from pyspark import SparkContext\n",
    "\n",
    "# Spark offers built-in packages to load CSV files\n",
    "sc =SparkContext()\n",
    "sqlContext = SQLContext(sc)\n",
    "data = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('crime_train.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Data Preprocessing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Remove the columns we do not need and have a look the first five rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "drop_list = ['Dates', 'DayOfWeek', 'PdDistrict', 'Resolution', 'Address', 'X', 'Y']\n",
    "data = data.select([column for column in data.columns if column not in drop_list])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Category: string (nullable = true)\n",
      " |-- Descript: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+--------------------+\n",
      "|      Category|            Descript|\n",
      "+--------------+--------------------+\n",
      "|      WARRANTS|      WARRANT ARREST|\n",
      "|OTHER OFFENSES|TRAFFIC VIOLATION...|\n",
      "|OTHER OFFENSES|TRAFFIC VIOLATION...|\n",
      "| LARCENY/THEFT|GRAND THEFT FROM ...|\n",
      "| LARCENY/THEFT|GRAND THEFT FROM ...|\n",
      "+--------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+\n",
      "|            Category| count|\n",
      "+--------------------+------+\n",
      "|       LARCENY/THEFT|174900|\n",
      "|      OTHER OFFENSES|126182|\n",
      "|        NON-CRIMINAL| 92304|\n",
      "|             ASSAULT| 76876|\n",
      "|       DRUG/NARCOTIC| 53971|\n",
      "|       VEHICLE THEFT| 53781|\n",
      "|           VANDALISM| 44725|\n",
      "|            WARRANTS| 42214|\n",
      "|            BURGLARY| 36755|\n",
      "|      SUSPICIOUS OCC| 31414|\n",
      "|      MISSING PERSON| 25989|\n",
      "|             ROBBERY| 23000|\n",
      "|               FRAUD| 16679|\n",
      "|FORGERY/COUNTERFE...| 10609|\n",
      "|     SECONDARY CODES|  9985|\n",
      "|         WEAPON LAWS|  8555|\n",
      "|        PROSTITUTION|  7484|\n",
      "|            TRESPASS|  7326|\n",
      "|     STOLEN PROPERTY|  4540|\n",
      "|SEX OFFENSES FORC...|  4388|\n",
      "+--------------------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Top 20 crimes\n",
    "from pyspark.sql.functions import col\n",
    "data.groupBy(\"Category\") \\\n",
    "    .count() \\\n",
    "    .orderBy(col(\"count\").desc()) \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|            Descript|count|\n",
      "+--------------------+-----+\n",
      "|GRAND THEFT FROM ...|60022|\n",
      "|       LOST PROPERTY|31729|\n",
      "|             BATTERY|27441|\n",
      "|   STOLEN AUTOMOBILE|26897|\n",
      "|DRIVERS LICENSE, ...|26839|\n",
      "|      WARRANT ARREST|23754|\n",
      "|SUSPICIOUS OCCURR...|21891|\n",
      "|AIDED CASE, MENTA...|21497|\n",
      "|PETTY THEFT FROM ...|19771|\n",
      "|MALICIOUS MISCHIE...|17789|\n",
      "|   TRAFFIC VIOLATION|16471|\n",
      "|PETTY THEFT OF PR...|16196|\n",
      "|MALICIOUS MISCHIE...|15957|\n",
      "|THREATS AGAINST LIFE|14716|\n",
      "|      FOUND PROPERTY|12146|\n",
      "|ENROUTE TO OUTSID...|11470|\n",
      "|GRAND THEFT OF PR...|11010|\n",
      "|POSSESSION OF NAR...|10050|\n",
      "|PETTY THEFT FROM ...|10029|\n",
      "|PETTY THEFT SHOPL...| 9571|\n",
      "+--------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Top 20 descriptions\n",
    "data.groupBy(\"Descript\") \\\n",
    "    .count() \\\n",
    "    .orderBy(col(\"count\").desc()) \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Create a Model\n",
    "### In Spark, we call create 'Model Pipeline', and we accomplish this in 5 steps\n",
    "1. regexTokenizer: Tokenization (with Regular Expression)\n",
    "2. stopwordsRemover: Remove Stop Words\n",
    "3. countVectors: Count vectors (“document-term vectors”)\n",
    "4. StringIndexer :  encodes a string column of labels to a column of label indices. The indices are in (0, numLabels), ordered by label frequencies, so the most frequent label gets index 0. \\\n",
    "(In our case, the label column (Category) will be encoded to label indices, from 0 to 32; the most frequent label (LARCENY/THEFT) will be indexed as 0)\n",
    "5. Split Train/Test data for making 'training-ready'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import RegexTokenizer, StopWordsRemover, CountVectorizer, OneHotEncoder, StringIndexer, VectorAssembler, HashingTF, IDF\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "# Clean description using regular-expression tokenizer\n",
    "regexTokenizer = RegexTokenizer(inputCol=\"Descript\", outputCol=\"words\", pattern=\"\\\\W\")\n",
    "\n",
    "# exclue stop words\n",
    "add_stopwords = [\"http\",\"https\",\"amp\",\"rt\",\"t\",\"c\",\"the\"] \n",
    "stopwordsRemover = StopWordsRemover(inputCol=\"words\", outputCol=\"filtered\").setStopWords(add_stopwords)\n",
    "\n",
    "# bag-of-words count\n",
    "countVectors = CountVectorizer(inputCol=\"filtered\", outputCol=\"features\", vocabSize=10000, minDF=5)\n",
    "\n",
    "# Add Hashing\n",
    "# hashingTF = HashingTF(inputCol=\"filtered\", outputCol=\"rawFeatures\", numFeatures=10000)\n",
    "\n",
    "# TF and IDF\n",
    "# idf = IDF(inputCol=\"rawFeatures\", outputCol=\"features\", minDocFreq=5) #minDocFreq: remove sparse terms\n",
    "\n",
    "# StringIndexer\n",
    "label_stringIdx = StringIndexer(inputCol = \"Category\", outputCol = \"label\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+--------------------+--------------------+--------------------+--------------------+--------------------+-----+\n",
      "|      Category|            Descript|               words|            filtered|         rawFeatures|            features|label|\n",
      "+--------------+--------------------+--------------------+--------------------+--------------------+--------------------+-----+\n",
      "|      WARRANTS|      WARRANT ARREST|   [warrant, arrest]|   [warrant, arrest]|(10000,[2279,3942...|(10000,[2279,3942...|  7.0|\n",
      "|OTHER OFFENSES|TRAFFIC VIOLATION...|[traffic, violati...|[traffic, violati...|(10000,[604,3942,...|(10000,[604,3942,...|  1.0|\n",
      "|OTHER OFFENSES|TRAFFIC VIOLATION...|[traffic, violati...|[traffic, violati...|(10000,[604,3942,...|(10000,[604,3942,...|  1.0|\n",
      "| LARCENY/THEFT|GRAND THEFT FROM ...|[grand, theft, fr...|[grand, theft, fr...|(10000,[274,713,3...|(10000,[274,713,3...|  0.0|\n",
      "| LARCENY/THEFT|GRAND THEFT FROM ...|[grand, theft, fr...|[grand, theft, fr...|(10000,[274,713,3...|(10000,[274,713,3...|  0.0|\n",
      "+--------------+--------------------+--------------------+--------------------+--------------------+--------------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml import Pipeline\n",
    "\n",
    "# Put everything in pipeline (We use regexTokenizer, stopwordsRemover, hashingTF, idf, label_stringIdx)\n",
    "# you can use hasingTF and IDF alternatively than countVectors\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": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Training Dataset Count: 658302\n",
      "Test Dataset Count: 219747\n"
     ]
    }
   ],
   "source": [
    "# Split Train/Test data\n",
    "(trainingData, testData) = dataset.randomSplit([0.75, 0.25], seed = 623)\n",
    "print(\"Training Dataset Count: \" + str(trainingData.count()))\n",
    "print(\"Test Dataset Count: \" + str(testData.count()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Train a Model and Evaluation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our model will make predictions and score on the test set \\\n",
    "And then we then look at the top 10 predictions from the highest probability."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------+-------------+------------------------------+-----+----------+\n",
      "|                      Descript|     Category|                   probability|label|prediction|\n",
      "+------------------------------+-------------+------------------------------+-----+----------+\n",
      "|THEFT, BICYCLE, <$50, NO SE...|LARCENY/THEFT|[0.8741040478278337,0.02016...|  0.0|       0.0|\n",
      "|THEFT, BICYCLE, <$50, NO SE...|LARCENY/THEFT|[0.8741040478278337,0.02016...|  0.0|       0.0|\n",
      "|THEFT, BICYCLE, <$50, NO SE...|LARCENY/THEFT|[0.8741040478278337,0.02016...|  0.0|       0.0|\n",
      "|THEFT, BICYCLE, <$50, NO SE...|LARCENY/THEFT|[0.8741040478278337,0.02016...|  0.0|       0.0|\n",
      "|THEFT, BICYCLE, <$50, NO SE...|LARCENY/THEFT|[0.8741040478278337,0.02016...|  0.0|       0.0|\n",
      "|THEFT, BICYCLE, <$50, NO SE...|LARCENY/THEFT|[0.8741040478278337,0.02016...|  0.0|       0.0|\n",
      "|THEFT, BICYCLE, <$50, NO SE...|LARCENY/THEFT|[0.8741040478278337,0.02016...|  0.0|       0.0|\n",
      "|THEFT, BICYCLE, <$50, SERIA...|LARCENY/THEFT|[0.874103341787826,0.020162...|  0.0|       0.0|\n",
      "|THEFT, BICYCLE, <$50, SERIA...|LARCENY/THEFT|[0.874103341787826,0.020162...|  0.0|       0.0|\n",
      "|THEFT, BICYCLE, <$50, SERIA...|LARCENY/THEFT|[0.874103341787826,0.020162...|  0.0|       0.0|\n",
      "+------------------------------+-------------+------------------------------+-----+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# We use Logistic-Regression model\n",
    "lr = LogisticRegression(maxIter=20, regParam=0.3, elasticNetParam=0)\n",
    "\n",
    "lrModel = lr.fit(trainingData)\n",
    "predictions = lrModel.transform(testData)\n",
    "predictions.filter(predictions['prediction'] == 0) \\\n",
    "    .select(\"Descript\",\"Category\",\"probability\",\"label\",\"prediction\") \\\n",
    "    .orderBy(\"probability\", ascending=False) \\\n",
    "    .show(n = 10, truncate = 30)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test-set Accuracy is :  0.972745626745252\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "evaluator = MulticlassClassificationEvaluator(predictionCol=\"prediction\")\n",
    "print(\"Test-set Accuracy is : \", evaluator.evaluate(predictions))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Cross-Validation (hyper-parameters tuning)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's try to improve our model by cross-validation, and we will tune the count vectors Logistic Regression"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Same Pipeline step like above\n",
    "pipeline = Pipeline(stages=[regexTokenizer, stopwordsRemover, countVectors, label_stringIdx])\n",
    "pipelineFit = pipeline.fit(data)\n",
    "dataset = pipelineFit.transform(data)\n",
    "(trainingData, testData) = dataset.randomSplit([0.75, 0.25], seed = 623)\n",
    "\n",
    "# Create LR model\n",
    "lr = LogisticRegression(maxIter=20, regParam=0.3, elasticNetParam=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "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",
    "cvModel = cv.fit(trainingData)\n",
    "\n",
    "predictions = cvModel.transform(testData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test-set Accuracy is :  0.9918902377560262\n"
     ]
    }
   ],
   "source": [
    "# Evaluate best model\n",
    "evaluator = MulticlassClassificationEvaluator(predictionCol=\"prediction\")\n",
    "print(\"Test-set Accuracy is : \", evaluator.evaluate(predictions))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We are able to acheive over 99% accuracy! "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Reference\n",
    ">https://towardsdatascience.com/multi-class-text-classification-with-pyspark-7d78d022ed35"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}