notebooks/samples/105 - Regression with DataConversion.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 105 - Training Regressions\n",
    "\n",
    "This example notebook is similar to\n",
    "[Notebook 102](102 - Regression Example with Flight Delay Dataset.ipynb).\n",
    "In this example, we will demonstrate the use of `DataConversion()` in two\n",
    "ways.  First, to convert the data type of several columns after the dataset\n",
    "has been read in to the Spark DataFrame instead of specifying the data types\n",
    "as the file is read in.  Second, to convert columns to categorical columns\n",
    "instead of iterating over the columns and applying the `StringIndexer`.\n",
    "\n",
    "This sample demonstrates how to use the following APIs:\n",
    "- [`TrainRegressor`\n",
    "  ](http://mmlspark.azureedge.net/docs/pyspark/TrainRegressor.html)\n",
    "- [`ComputePerInstanceStatistics`\n",
    "  ](http://mmlspark.azureedge.net/docs/pyspark/ComputePerInstanceStatistics.html)\n",
    "- [`DataConversion`\n",
    "  ](http://mmlspark.azureedge.net/docs/pyspark/DataConversion.html)\n",
    "\n",
    "First, import the pandas package"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, import the CSV dataset: retrieve the file if needed, save it locally,\n",
    "read the data into a pandas dataframe via `read_csv()`, then convert it to\n",
    "a Spark dataframe.\n",
    "\n",
    "Print the schema of the dataframe, and note the columns that are `long`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataFile = \"On_Time_Performance_2012_9.csv\"\n",
    "import os, urllib\n",
    "if not os.path.isfile(dataFile):\n",
    "    urllib.request.urlretrieve(\"https://mmlspark.azureedge.net/datasets/\"+dataFile, dataFile)\n",
    "flightDelay = spark.createDataFrame(pd.read_csv(dataFile))\n",
    "# print some basic info\n",
    "print(\"records read: \" + str(flightDelay.count()))\n",
    "print(\"Schema: \")\n",
    "flightDelay.printSchema()\n",
    "flightDelay.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `DataConversion` transform API to convert the columns listed to\n",
    "double.\n",
    "\n",
    "The `DataConversion` API accepts the following types for the `convertTo`\n",
    "parameter:\n",
    "* `boolean`\n",
    "* `byte`\n",
    "* `short`\n",
    "* `integer`\n",
    "* `long`\n",
    "* `float`\n",
    "* `double`\n",
    "* `string`\n",
    "* `toCategorical`\n",
    "* `clearCategorical`\n",
    "* `date` -- converts a string or long to a date of the format\n",
    "  \"yyyy-MM-dd HH:mm:ss\" unless another format is specified by\n",
    "the `dateTimeFormat` parameter.\n",
    "\n",
    "Again, print the schema and note that the columns are now `double`\n",
    "instead of long."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mmlspark import DataConversion\n",
    "flightDelay = DataConversion(col=\"Quarter,Month,DayofMonth,DayOfWeek,\"\n",
    "                                 + \"OriginAirportID,DestAirportID,\"\n",
    "                                 + \"CRSDepTime,CRSArrTime\",\n",
    "                             convertTo=\"double\") \\\n",
    "                  .transform(flightDelay)\n",
    "flightDelay.printSchema()\n",
    "flightDelay.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Split the datasest into train and test sets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train, test = flightDelay.randomSplit([0.75, 0.25])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a regressor model and train it on the dataset.\n",
    "\n",
    "First, use `DataConversion` to convert the columns `Carrier`, `DepTimeBlk`,\n",
    "and `ArrTimeBlk` to categorical data.  Recall that in Notebook 102, this\n",
    "was accomplished by iterating over the columns and converting the strings\n",
    "to index values using the `StringIndexer` API.  The `DataConversion` API\n",
    "simplifies the task by allowing you to specify all columns that will have\n",
    "the same end type in a single command.\n",
    "\n",
    "Create a LinearRegression model using the Limited-memory BFGS solver\n",
    "(`l-bfgs`), an `ElasticNet` mixing parameter of `0.3`, and a `Regularization`\n",
    "of `0.1`.\n",
    "\n",
    "Train the model with the `TrainRegressor` API fit on the training dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mmlspark import TrainRegressor, TrainedRegressorModel\n",
    "from pyspark.ml.regression import LinearRegression\n",
    "\n",
    "trainCat = DataConversion(col=\"Carrier,DepTimeBlk,ArrTimeBlk\",\n",
    "                          convertTo=\"toCategorical\") \\\n",
    "               .transform(train)\n",
    "testCat  = DataConversion(col=\"Carrier,DepTimeBlk,ArrTimeBlk\",\n",
    "                          convertTo=\"toCategorical\") \\\n",
    "               .transform(test)\n",
    "lr = LinearRegression().setSolver(\"l-bfgs\").setRegParam(0.1) \\\n",
    "                       .setElasticNetParam(0.3)\n",
    "model = TrainRegressor(model=lr, labelCol=\"ArrDelay\").fit(trainCat)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Score the regressor on the test data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "scoredData = model.transform(testCat)\n",
    "scoredData.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Compute model metrics against the entire scored dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mmlspark import ComputeModelStatistics\n",
    "metrics = ComputeModelStatistics().transform(scoredData)\n",
    "metrics.toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, compute and show statistics on individual predictions in the test\n",
    "dataset, demonstrating the usage of `ComputePerInstanceStatistics`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mmlspark import ComputePerInstanceStatistics\n",
    "evalPerInstance = ComputePerInstanceStatistics().transform(scoredData)\n",
    "evalPerInstance.select(\"ArrDelay\", \"Scores\", \"L1_loss\", \"L2_loss\") \\\n",
    "               .limit(10).toPandas()"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [default]",
   "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}