notebooks/samples/Classification - Twitter Sentiment with Vowpal Wabbit.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Twitter Sentiment Classification using Vowpal Wabbit in MMLSpark\n",
    "\n",
    "In this example, we show how to build a sentiment classification model using Vowpal Wabbit (VW) in MMLSpark. The data set we use to train and evaluate the model is [Sentiment140](http://help.sentiment140.com/for-students/?source=post_page---------------------------) twitter data. First, we import a few packages that we need."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import re\n",
    "import urllib.request\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from zipfile import ZipFile\n",
    "from bs4 import BeautifulSoup\n",
    "from pyspark.sql.functions import udf, rand, when, col\n",
    "from pyspark.sql.types import StructType, StructField, DoubleType, StringType\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import CountVectorizer, RegexTokenizer\n",
    "from mmlspark.vw import VowpalWabbitClassifier\n",
    "from mmlspark.train import ComputeModelStatistics\n",
    "from pyspark.mllib.evaluation import BinaryClassificationMetrics\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# URL to download the sentiment140 dataset and data file names\n",
    "DATA_URL = \"http://cs.stanford.edu/people/alecmgo/trainingandtestdata.zip\"\n",
    "TRAIN_FILENAME = \"training.1600000.processed.noemoticon.csv\"\n",
    "TEST_FILENAME = \"testdata.manual.2009.06.14.csv\"\n",
    "# Folder for storing the downloaded data\n",
    "DATA_FOLDER = \"data\"\n",
    "# Data column names\n",
    "COL_NAMES = [\"label\", \"id\", \"date\", \"query_string\", \"user\", \"text\"]\n",
    "# Text encoding type of the data\n",
    "ENCODING = \"iso-8859-1\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Preparation\n",
    "\n",
    "We use [Sentiment140](http://help.sentiment140.com/for-students/?source=post_page---------------------------) twitter data which originated from a Standford research project to train and evaluate VW classification model on Spark. The same dataset has been used in a previous [Azure Machine Learning sample](https://github.com/Azure-Samples/MachineLearningSamples-TwitterSentimentPrediction) on twitter sentiment prediction. Before using the data to build the classification model, we first download and clean up the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def download_data(url, data_folder=DATA_FOLDER, filename=\"downloaded_data.zip\"):\n",
    "    \"\"\"Download and extract data from url\"\"\"\n",
    "    \n",
    "    data_dir = \"./\" + DATA_FOLDER\n",
    "    if not os.path.exists(data_dir): os.makedirs(data_dir)\n",
    "    downloaded_filepath = os.path.join(data_dir, filename)\n",
    "    print(\"Downloading data...\")\n",
    "    urllib.request.urlretrieve(url, downloaded_filepath)\n",
    "    print(\"Extracting data...\")\n",
    "    zipfile = ZipFile(downloaded_filepath)\n",
    "    zipfile.extractall(data_dir)\n",
    "    zipfile.close()\n",
    "    print(\"Finished data downloading and extraction.\")\n",
    "    \n",
    "download_data(DATA_URL)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's read the training data into a Spark DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_train = pd.read_csv(os.path.join(\".\", DATA_FOLDER, TRAIN_FILENAME), \n",
    "                       header=None, names=COL_NAMES, encoding=ENCODING)\n",
    "df_train = spark.createDataFrame(df_train, verifySchema=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can take a look at the training data and check how many samples it has. We should see that there are 1.6 million samples in the training data. There are 6 fields in the training data:\n",
    "* label: the sentiment of the tweet (0.0 = negative, 2.0 = neutral, 4.0 = positive)\n",
    "* id: the id of the tweet\n",
    "* date: the date of the tweet\n",
    "* query_string: The query used to extract the data. If there is no query, then this value is NO_QUERY.\n",
    "* user: the user that tweeted\n",
    "* text: the text of the tweet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_train.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Number of training samples: \", df_train.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Before training the model, we randomly permute the data to mix negative and positive samples. This is helpful for properly training online learning algorithms like VW. To speed up model training, we use a subset of the data to train the model. If training with the full training set, typically you will see better performance of the model on the test set. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_train = df_train.orderBy(rand()) \\\n",
    "                   .limit(100000) \\\n",
    "                   .withColumn(\"label\", when(col(\"label\") > 0, 1.0).otherwise(0.0)) \\\n",
    "                   .select([\"label\", \"text\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## VW MMLSpark Training\n",
    "\n",
    "Now we are ready to define a pipeline which consists of feture engineering steps and the VW model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Specify featurizers\n",
    "tokenizer = RegexTokenizer(inputCol=\"text\",\n",
    "                           outputCol=\"words\")\n",
    "\n",
    "count_vectorizer = CountVectorizer(inputCol=\"words\",\n",
    "                                   outputCol=\"features\")\n",
    "\n",
    "# Define VW classification model\n",
    "args = \"--loss_function=logistic --quiet --holdout_off\"\n",
    "vw_model = VowpalWabbitClassifier(featuresCol=\"features\", \n",
    "                                  labelCol=\"label\", \n",
    "                                  args=args, \n",
    "                                  numPasses=10)\n",
    "\n",
    "# Create a pipeline\n",
    "vw_pipeline = Pipeline(stages=[tokenizer, count_vectorizer, vw_model])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With the prepared training data, we can fit the model pipeline as follows."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "vw_trained = vw_pipeline.fit(df_train)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model Performance Evaluation\n",
    "\n",
    "After training the model, we evaluate the performance of the model using the test set which is manually labeled."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_test = pd.read_csv(os.path.join(\".\", DATA_FOLDER, TEST_FILENAME), \n",
    "                       header=None, names=COL_NAMES, encoding=ENCODING)\n",
    "df_test = spark.createDataFrame(df_test, verifySchema=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We only use positive and negative tweets in the test set to evaluate the model, since our model is a binary classification model trained with only positive and negative tweets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Number of test samples before filtering: \", df_test.count())\n",
    "df_test = df_test.filter(col(\"label\") != 2.0) \\\n",
    "                 .withColumn(\"label\", when(col(\"label\") > 0, 1.0).otherwise(0.0)) \\\n",
    "                 .select([\"label\", \"text\"])\n",
    "print(\"Number of test samples after filtering: \", df_test.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Make predictions\n",
    "predictions = vw_trained.transform(df_test)\n",
    "predictions.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute model performance metrics\n",
    "metrics = ComputeModelStatistics(evaluationMetric=\"classification\", \n",
    "                                 labelCol=\"label\", \n",
    "                                 scoredLabelsCol=\"prediction\").transform(predictions)\n",
    "metrics.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Utility class for plotting ROC curve (https://stackoverflow.com/questions/52847408/pyspark-extract-roc-curve)\n",
    "class CurveMetrics(BinaryClassificationMetrics):\n",
    "    def __init__(self, *args):\n",
    "        super(CurveMetrics, self).__init__(*args)\n",
    "\n",
    "    def get_curve(self, method):\n",
    "        rdd = getattr(self._java_model, method)().toJavaRDD()\n",
    "        points = []\n",
    "        for row in rdd.collect():\n",
    "            points += [(float(row._1()), float(row._2()))]\n",
    "        return points\n",
    "\n",
    "preds = predictions.select(\"label\", \"probability\") \\\n",
    "                   .rdd.map(lambda row: (float(row[\"probability\"][1]), float(row[\"label\"])))\n",
    "roc_points = CurveMetrics(preds).get_curve(\"roc\")\n",
    "\n",
    "# Plot ROC curve\n",
    "fig = plt.figure()\n",
    "x_val = [x[0] for x in roc_points]\n",
    "y_val = [x[1] for x in roc_points]\n",
    "plt.title(\"ROC curve on test set\")\n",
    "plt.xlabel(\"False positive rate\")\n",
    "plt.ylabel(\"True positive rate\")\n",
    "plt.plot(x_val, y_val)\n",
    "# Use display() if you're on Azure Databricks or you can do plt.show()\n",
    "display(fig)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You should see an ROC curve like the following after the above cell is executed. \n",
    "\n",
    "<img src=\"https://user-images.githubusercontent.com/20047467/69376052-9b0a3380-0c77-11ea-9266-11aa44350cbe.png\" width=\"400\" height=\"320\">"
   ]
  }
 ],
 "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.6.8"
  },
  "name": "vw_mmlspark_sentiment_classification2",
  "notebookId": 2916790739696591
 },
 "nbformat": 4,
 "nbformat_minor": 1
}