Spark ML Pipelines and MLeap Demo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Spark ML Pipelines - Lending Club Demo"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Using the Lending Club data set, this demo will demonstrate how to:\n",
    "\n",
    "1. Load the research dataset from s3\n",
    "2. Perform feature extraction using Spark ML APIs\n",
    "3. Train a Logistic Regression and a Random Forest Classifier\n",
    "4. Perform Hyperparameter tuning using ParamGrid, Binary Classification Evaluator, and a Cross Validator\n",
    "5. Export the best pipeline (feature transformers and LR/RF models) to an MLeap Bundle, which we'll use to deploy the pipeline to an RESTful API service.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Cluster Configuration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>84</td><td>application_1497282488964_0084</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-10-30-1-122.us-west-2.compute.internal:20888/proxy/application_1497282488964_0084/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-10-30-1-118.us-west-2.compute.internal:8042/node/containerlogs/container_1497282488964_0084_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "Current session configs: <tt>{'conf': {'spark.jars.packages': 'org.apache.hadoop:hadoop-aws:2.7.3,ml.combust.mleap:mleap-spark_2.11:0.7.0,com.databricks:spark-avro_2.11:3.0.1', 'spark.jars.excludes': 'org.scala-lang:scala-reflect,org.apache.spark:spark-tags_2.11'}, 'driverMemory': '2048M', 'kind': 'spark', 'executorCores': 2}</tt><br>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>81</td><td>application_1497282488964_0081</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-10-30-1-122.us-west-2.compute.internal:20888/proxy/application_1497282488964_0081/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-10-30-1-154.us-west-2.compute.internal:8042/node/containerlogs/container_1497282488964_0081_01_000001/root\">Link</a></td><td></td></tr><tr><td>82</td><td>application_1497282488964_0082</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-10-30-1-122.us-west-2.compute.internal:20888/proxy/application_1497282488964_0082/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-10-30-1-165.us-west-2.compute.internal:8042/node/containerlogs/container_1497282488964_0082_01_000001/root\">Link</a></td><td></td></tr><tr><td>84</td><td>application_1497282488964_0084</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-10-30-1-122.us-west-2.compute.internal:20888/proxy/application_1497282488964_0084/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-10-30-1-118.us-west-2.compute.internal:8042/node/containerlogs/container_1497282488964_0084_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%configure -f\n",
    "{\"kind\": \"spark\",\n",
    "\"driverMemory\": \"2048M\",\n",
    "\"executorCores\": 2,\n",
    " \"conf\":{\"spark.jars.packages\":\"org.apache.hadoop:hadoop-aws:2.7.3,ml.combust.mleap:mleap-spark_2.11:0.7.0,com.databricks:spark-avro_2.11:3.0.1\", \"spark.jars.excludes\": \"org.scala-lang:scala-reflect,org.apache.spark:spark-tags_2.11\"}\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Part 2: Spark ML Feature Extraction"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Load Required Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>80</td><td>application_1497282488964_0080</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-10-30-1-122.us-west-2.compute.internal:20888/proxy/application_1497282488964_0080/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-10-30-1-165.us-west-2.compute.internal:8042/node/containerlogs/container_1497282488964_0080_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n",
      "import org.apache.spark.ml.bundle.SparkBundleContext"
     ]
    }
   ],
   "source": [
    "// Spark Training Pipeline Libraries\n",
    "import org.apache.spark.ml.feature.OneHotEncoder\n",
    "import org.apache.spark.ml.feature.{StandardScaler, StringIndexer, VectorAssembler, PolynomialExpansion}\n",
    "import org.apache.spark.ml.classification.{RandomForestClassifier, LogisticRegression, RandomForestClassificationModel}\n",
    "import org.apache.spark.ml.{Pipeline, PipelineModel, Transformer, PipelineStage}\n",
    "import org.apache.spark.ml.tuning.{CrossValidator, ParamGridBuilder}\n",
    "import org.apache.spark.ml.evaluation.{BinaryClassificationEvaluator}\n",
    "import com.databricks.spark.avro._\n",
    "\n",
    "// MLeap/Bundle.ML Serialization Libraries\n",
    "import ml.combust.mleap.spark.SparkSupport._\n",
    "import resource._\n",
    "import ml.combust.bundle.BundleFile\n",
    "import org.apache.spark.ml.bundle.SparkBundleContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 1: Import and Explore the Research Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total N Records: 797987"
     ]
    }
   ],
   "source": [
    "//val filePath = \"s3a://mleap-demo/datasources/lending_club_2017060401.avro\"\n",
    "val filePath = \"hdfs:///data/lending_club_2017060401.avro\"\n",
    "val dataset = spark.read.avro(filePath)\n",
    "\n",
    "println(\"Total N Records: \" + dataset.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total N Approved Records: 42535"
     ]
    }
   ],
   "source": [
    "println(\"Total N Approved Records: \" + dataset.filter(\"approved == 1.0\").count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------------------+-----+----------+-----+--------+-------------+\n",
      "|loan_amount|fico_score_group_fnl|  dti|emp_length|state|approved|   loan_title|\n",
      "+-----------+--------------------+-----+----------+-----+--------+-------------+\n",
      "|     5000.0|           650 - 700|27.65| 10+ years|   AZ|     1.0|  Electronics|\n",
      "|     2500.0|           600 - 650|  1.0|  < 1 year|   GA|     1.0|   Motorcycle|\n",
      "|     2400.0|           600 - 650| 8.72| 10+ years|   IL|     1.0|Business Loan|\n",
      "|    10000.0|           600 - 650| 20.0| 10+ years|   CA|     1.0|Personal Loan|\n",
      "|     3000.0|           650 - 700|17.94|    1 year|   OR|     1.0|Personal Loan|\n",
      "+-----------+--------------------+-----+----------+-----+--------+-------------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "// View the individual records\n",
    "dataset.select(\"loan_amount\", \"fico_score_group_fnl\", \"dti\", \"emp_length\", \"state\", \"approved\", \"loan_title\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Alter the Data Slightly For The Demo\n",
    "\n",
    "- Cap DTI\n",
    "- Limit the number of states for OneHotEncoding\n",
    "- Keep only certain loan categories"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "796751"
     ]
    }
   ],
   "source": [
    "dataset.registerTempTable(\"df\")\n",
    "println(dataset.count())\n",
    "\n",
    "val datasetFnl = spark.sqlContext.sql(f\"\"\"\n",
    "    select\n",
    "        loan_amount,\n",
    "        fico_score_group_fnl,\n",
    "        case when dti >= 10.0\n",
    "            then 10.0\n",
    "            else dti\n",
    "        end as dti,\n",
    "        emp_length,\n",
    "        case when state in ('CA', 'NY', 'MN', 'IL', 'FL', 'WA', 'MA', 'TX', 'GA', 'OH', 'NJ', 'VA', 'MI')\n",
    "            then state\n",
    "            else 'Other'\n",
    "        end as state,\n",
    "        loan_title,\n",
    "        approved\n",
    "    from df\n",
    "    where loan_title in('Debt Consolidation', 'Other', 'Home/Home Improvement', 'Payoff Credit Card', 'Car Payment/Loan',\n",
    "    'Business Loan', 'Health/Medical', 'Moving', 'Wedding/Engagement', 'Vacation', 'College', 'Renewable Energy', 'Payoff Bills',\n",
    "    'Personal Loan', 'Motorcycle')\n",
    "\"\"\")\n",
    "\n",
    "datasetFnl.registerTempTable(\"dfFnl\")\n",
    "println(datasetFnl.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Summary Statistics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------+-----------+----+--------+\n",
      "|state|     n|loan_amount| dti|approved|\n",
      "+-----+------+-----------+----+--------+\n",
      "|   MA| 20217|   12371.90|0.86|0.070782|\n",
      "|   CA|107084|   13524.24|0.87|0.069021|\n",
      "|   NJ| 29613|   13781.77|0.84|0.066761|\n",
      "|   NY| 64701|   13120.19|0.79|0.062565|\n",
      "|   WA| 14878|   13517.80|0.80|0.059282|\n",
      "|   VA| 25018|   12865.42|0.87|0.059197|\n",
      "|   MN| 11036|   12790.27|0.80|0.058536|\n",
      "|   IL| 33109|   13138.97|0.73|0.050198|\n",
      "|   FL| 63016|   12390.05|0.70|0.048543|\n",
      "|   GA| 30428|   12307.86|0.74|0.048541|\n",
      "|Other|281543|   12508.86|0.70|0.046622|\n",
      "|   TX| 64858|   13090.81|0.73|0.044466|\n",
      "|   OH| 29796|   12094.84|0.71|0.044234|\n",
      "|   MI| 21454|   12582.70|0.62|0.036590|\n",
      "+-----+------+-----------+----+--------+"
     ]
    }
   ],
   "source": [
    "// Most popular cities (original dataset)\n",
    "\n",
    "spark.sqlContext.sql(f\"\"\"\n",
    "    select \n",
    "        state,\n",
    "        count(*) as n,\n",
    "        cast(avg(loan_amount) as decimal(12,2)) as loan_amount,\n",
    "        cast(avg(dti) as decimal(12,2)) as dti,\n",
    "        avg(cast(approved as decimal(12,2))) as approved\n",
    "    from dfFnl\n",
    "    group by state\n",
    "    order by avg(cast(approved as decimal(12,2))) desc\n",
    "\"\"\").show(15)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+----+--------+\n",
      "|          loan_title|     n|loan_amount| dti|approved|\n",
      "+--------------------+------+-----------+----+--------+\n",
      "|          Motorcycle|   732|    7059.05|2.49|0.289617|\n",
      "|        Payoff Bills|  3162|   10763.05|2.61|0.271980|\n",
      "|       Personal Loan|  2981|    9366.92|2.24|0.243207|\n",
      "|  Debt Consolidation|315519|   14769.15|0.90|0.070924|\n",
      "|               Other|190221|    9442.56|0.86|0.054605|\n",
      "|  Wedding/Engagement| 13568|   10148.23|0.63|0.051739|\n",
      "|             College|  7989|    7923.65|0.83|0.044812|\n",
      "|  Payoff Credit Card| 60197|   15821.83|0.70|0.039387|\n",
      "|Home/Home Improve...| 60779|   15176.65|0.42|0.028085|\n",
      "|      Health/Medical| 20615|    7470.27|0.43|0.023575|\n",
      "|    Car Payment/Loan| 47387|   10308.56|0.38|0.021504|\n",
      "|            Vacation|  9905|    5626.50|0.38|0.017769|\n",
      "|              Moving| 18733|    6621.09|0.38|0.014573|\n",
      "|       Business Loan| 41792|   18151.20|0.36|0.012132|\n",
      "|    Renewable Energy|  3171|    9786.62|0.34|0.001892|\n",
      "+--------------------+------+-----------+----+--------+"
     ]
    }
   ],
   "source": [
    "// Most popular cities (original dataset)\n",
    "\n",
    "spark.sqlContext.sql(f\"\"\"\n",
    "    select \n",
    "        loan_title,\n",
    "        count(*) as n,\n",
    "        cast(avg(loan_amount) as decimal(12,2)) as loan_amount,\n",
    "        cast(avg(dti) as decimal(12,2)) as dti,\n",
    "        avg(cast(approved as decimal(12,2))) as approved\n",
    "    from dfFnl\n",
    "    group by loan_title\n",
    "    order by avg(cast(approved as decimal(12,2))) desc\n",
    "\"\"\").show(15)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 2: Define continous and categorical features and filter nulls"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "allFeatures: Array[String] = Array(loan_amount, dti, loan_title, emp_length, state, fico_score_group_fnl)"
     ]
    }
   ],
   "source": [
    "// Step 2. Create our feature pipeline and train it on the entire dataset\n",
    "val continuousFeatures = Array(\"loan_amount\", \"dti\")\n",
    "\n",
    "val categoricalFeatures = Array(\"loan_title\",\n",
    "  \"emp_length\",\n",
    "  \"state\",\n",
    "  \"fico_score_group_fnl\")\n",
    "\n",
    "val allFeatures = continuousFeatures.union(categoricalFeatures)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "796751"
     ]
    }
   ],
   "source": [
    "// Filter all null values\n",
    "val allCols = allFeatures.union(Seq(\"approved\")).map(datasetFnl.col)\n",
    "val nullFilter = allCols.map(_.isNotNull).reduce(_ && _)\n",
    "val datasetImputedFiltered = datasetFnl.select(allCols: _*).filter(nullFilter).persist()\n",
    "\n",
    "println(datasetImputedFiltered.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 3: Split data into training and validation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "trainingDataset: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [loan_amount: double, dti: double ... 5 more fields]\n",
      "validationDataset: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [loan_amount: double, dti: double ... 5 more fields]"
     ]
    }
   ],
   "source": [
    "val Array(trainingDataset, validationDataset) = datasetImputedFiltered.randomSplit(Array(0.7, 0.3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 4: Continous Feature Pipelines\n",
    "\n",
    "Pipeline 1: Scale Features\n",
    "    - VectorAssembler\n",
    "    - StandardScaler\n",
    "    \n",
    "Pipeline 2: Polynomial Expansion\n",
    "    - VectorAssembler\n",
    "    - PolynomialExpansion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "continuousFeaturePolynomialExpansion: org.apache.spark.ml.feature.PolynomialExpansion = polynomial_expansion_loan_amount"
     ]
    }
   ],
   "source": [
    "// Pipeline 1\n",
    "val continuousFeatureAssembler = new VectorAssembler(uid = \"continuous_feature_assembler\").\n",
    "    setInputCols(continuousFeatures).\n",
    "    setOutputCol(\"unscaled_continuous_features\")\n",
    "\n",
    "val continuousFeatureScaler = new StandardScaler(uid = \"continuous_feature_scaler\").\n",
    "    setInputCol(\"unscaled_continuous_features\").\n",
    "    setOutputCol(\"scaled_continuous_features\")\n",
    "\n",
    "// Pipeline 2\n",
    "val polyExpansionAssembler = new VectorAssembler(uid = \"poly_expansion_feature_assembler\").\n",
    "    setInputCols(Array(\"loan_amount\", \"dti\")).\n",
    "    setOutputCol(\"poly_expansions_features\")\n",
    "\n",
    "val continuousFeaturePolynomialExpansion = new PolynomialExpansion(uid = \"polynomial_expansion_loan_amount\").\n",
    "    setInputCol(\"poly_expansions_features\").\n",
    "    setOutputCol(\"loan_amount_polynomial_expansion_features\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 5: Categorical Feature Pipeline\n",
    "\n",
    "Pipeline 3: String Index + One-Hot-Encode each categorical feature\n",
    "    - StringIndexer\n",
    "    - OneHotEncoder"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "categoricalFeatureOneHotEncoders: Array[org.apache.spark.ml.feature.OneHotEncoder] = Array(oh_encoder_loan_title_index, oh_encoder_emp_length_index, oh_encoder_state_index, oh_encoder_fico_score_group_fnl_index)"
     ]
    }
   ],
   "source": [
    "val categoricalFeatureIndexers = categoricalFeatures.map {\n",
    "    feature => new StringIndexer(uid = s\"string_indexer_$feature\").\n",
    "      setInputCol(feature).\n",
    "      setOutputCol(s\"${feature}_index\")\n",
    "}\n",
    "\n",
    "val categoricalFeatureOneHotEncoders = categoricalFeatureIndexers.map {\n",
    "    indexer => new OneHotEncoder(uid = s\"oh_encoder_${indexer.getOutputCol}\").\n",
    "      setInputCol(indexer.getOutputCol).\n",
    "      setOutputCol(s\"${indexer.getOutputCol}_oh\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 6: Assemble our features and feature pipeline\n",
    "\n",
    "Combine Pipeline 1, Pipeline 2, and Pipeline 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finished constructing the pipeline"
     ]
    }
   ],
   "source": [
    "// Assemble Feature Vector For Random Forest Classifier\n",
    "val featureColsRf = categoricalFeatureIndexers.map(_.getOutputCol).union(Seq(\"scaled_continuous_features\", \"loan_amount_polynomial_expansion_features\"))\n",
    "\n",
    "// Assemble Feature Vector For Logistic Regression\n",
    "val featureColsLr = categoricalFeatureOneHotEncoders.map(_.getOutputCol).union(Seq(\"scaled_continuous_features\"))\n",
    "\n",
    "// Vector-assemble all categorical and continuous features into a single feature vector\n",
    "val featureAssemblerLr = new VectorAssembler(uid = \"feature_assembler_lr\").\n",
    "    setInputCols(featureColsLr).\n",
    "    setOutputCol(\"features_lr\")\n",
    "    \n",
    "val featureAssemblerRf = new VectorAssembler(uid = \"feature_assembler_rf\").\n",
    "    setInputCols(featureColsRf).\n",
    "    setOutputCol(\"features_rf\")\n",
    "\n",
    "// Define an array of all of our feature estimators that we need to fit\n",
    "val estimators: Array[PipelineStage] = Array(continuousFeatureAssembler, continuousFeatureScaler, polyExpansionAssembler, continuousFeaturePolynomialExpansion).\n",
    "    union(categoricalFeatureIndexers).\n",
    "    union(categoricalFeatureOneHotEncoders).\n",
    "    union(Seq(featureAssemblerLr, featureAssemblerRf))\n",
    "\n",
    "// Build our pipeline based on the estimators\n",
    "val featurePipeline = new Pipeline(uid = \"feature_pipeline\").\n",
    "    setStages(estimators)\n",
    "\n",
    "// Fit our pipeline\n",
    "val sparkFeaturePipelineModel = featurePipeline.fit(trainingDataset)\n",
    "\n",
    "println(\"Finished constructing the pipeline\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res64: Any = (46,[0,20,25,42,44,45],[1.0,1.0,1.0,1.0,0.04701321420936851,-0.004627647986286722])"
     ]
    }
   ],
   "source": [
    "val dfWithFeatures = sparkFeaturePipelineModel.transform(trainingDataset)\n",
    "// Explore our generated features - Linear Regression\n",
    "dfWithFeatures.select(\"features_lr\").head(1).head.get(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res51: Array[org.apache.spark.sql.Row] = Array([[0.0,6.0,0.0,4.0,0.04701321420936851,-0.004627647986286722,500.0,250000.0,-0.01,-5.0,1.0E-4]])"
     ]
    }
   ],
   "source": [
    "// Explore our generated features - Linear Regression\n",
    "dfWithFeatures.select(\"features_rf\").head(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Part 3: Spark ML Training and Scoring"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "In this section we are going to train a Logistic Regression model and a Random Forest Classifier model on the training data and test their accuracy on the validation dataset."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 7: Train Random Forest Classifier"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Complete: Training Random Forest"
     ]
    }
   ],
   "source": [
    "// Step 3.1 Create our random forest model\n",
    "val randomForest = new RandomForestClassifier(uid = \"random_forest_classifier\").\n",
    "    setFeaturesCol(\"features_rf\").\n",
    "    setLabelCol(\"approved\").\n",
    "    setPredictionCol(\"approved_prediction\").\n",
    "    setNumTrees(5).\n",
    "    setMaxDepth(5)\n",
    "\n",
    "val sparkPipelineEstimatorRf = new Pipeline().setStages(Array(sparkFeaturePipelineModel, randomForest))\n",
    "\n",
    "val sparkPipelineRf = sparkPipelineEstimatorRf.fit(trainingDataset)\n",
    "\n",
    "println(\"Complete: Training Random Forest\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res69: String =\n",
      "\"RandomForestClassificationModel (uid=rfc_9df1a346bde4) with 5 trees\n",
      "  Tree 0 (weight 1.0):\n",
      "    If (feature 10 <= 7.617599999999999)\n",
      "     If (feature 10 <= 0.4489000000000001)\n",
      "      If (feature 1 in {2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0})\n",
      "       If (feature 3 in {4.0,6.0})\n",
      "        If (feature 1 in {2.0,3.0,4.0,5.0,6.0,8.0,9.0,10.0,11.0})\n",
      "         Predict: 0.0\n",
      "        Else (feature 1 not in {2.0,3.0,4.0,5.0,6.0,8.0,9.0,10.0,11.0})\n",
      "         Predict: 0.0\n",
      "       Else (feature 3 not in {4.0,6.0})\n",
      "        If (feature 7 <= 9.9500625E7)\n",
      "         Predict: 0.0\n",
      "        Else (feature 7 > 9.9500625E7)\n",
      "         Predict: 0.0\n",
      "      Else (feature 1 not in {2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0})\n",
      "       If (feature 5 <= 0.20236704644031833)\n",
      "        If (feature 1 in {0.0})\n",
      "       ..."
     ]
    }
   ],
   "source": [
    "val randomForestModel = randomForest.fit(dfWithFeatures)\n",
    "randomForestModel.toDebugString"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res70: org.apache.spark.ml.param.ParamMap =\n",
      "{\n",
      "\trfc_9df1a346bde4-cacheNodeIds: false,\n",
      "\trfc_9df1a346bde4-checkpointInterval: 10,\n",
      "\trfc_9df1a346bde4-featureSubsetStrategy: auto,\n",
      "\trfc_9df1a346bde4-featuresCol: features_rf,\n",
      "\trfc_9df1a346bde4-impurity: gini,\n",
      "\trfc_9df1a346bde4-labelCol: approved,\n",
      "\trfc_9df1a346bde4-maxBins: 32,\n",
      "\trfc_9df1a346bde4-maxDepth: 5,\n",
      "\trfc_9df1a346bde4-maxMemoryInMB: 256,\n",
      "\trfc_9df1a346bde4-minInfoGain: 0.0,\n",
      "\trfc_9df1a346bde4-minInstancesPerNode: 1,\n",
      "\trfc_9df1a346bde4-numTrees: 5,\n",
      "\trfc_9df1a346bde4-predictionCol: approved_prediction,\n",
      "\trfc_9df1a346bde4-probabilityCol: probability,\n",
      "\trfc_9df1a346bde4-rawPredictionCol: rawPrediction,\n",
      "\trfc_9df1a346bde4-seed: 207336481,\n",
      "\trfc_9df1a346bde4-subsamplingRate: 1.0\n",
      "}"
     ]
    }
   ],
   "source": [
    "randomForestModel.extractParamMap"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 8: Train Logistic Regression Model\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Complete: Training Logistic Regression"
     ]
    }
   ],
   "source": [
    "val logisticRegression = new LogisticRegression(uid = \"logistic_regression\").\n",
    "    setFeaturesCol(\"features_lr\").\n",
    "    setLabelCol(\"approved\").\n",
    "    setPredictionCol(\"approved_prediction\")\n",
    "\n",
    "val sparkPipelineEstimatorLr = new Pipeline().setStages(Array(sparkFeaturePipelineModel, logisticRegression))\n",
    "\n",
    "val sparkPipelineLr = sparkPipelineEstimatorLr.fit(trainingDataset)\n",
    "\n",
    "println(\"Complete: Training Logistic Regression\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 9: Validate the Random Forest Classifier"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "validationDataWithPrediction: org.apache.spark.sql.DataFrame = [loan_amount: double, dti: double ... 22 more fields]"
     ]
    }
   ],
   "source": [
    "// Run the feature transformers and score the validation dataset with RF model\n",
    "val validationDataWithPrediction = sparkPipelineRf.transform(validationDataset)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy: 0.995613741352638"
     ]
    }
   ],
   "source": [
    "// Compute the Area Under the Receiver Operating Characteristics (ROC) Curve\n",
    "val evaluator = new BinaryClassificationEvaluator().\n",
    "    setLabelCol(randomForest.getLabelCol).\n",
    "    setRawPredictionCol(randomForest.getRawPredictionCol).\n",
    "    setMetricName(\"areaUnderROC\")\n",
    "    \n",
    "val accuracy = evaluator.evaluate(validationDataWithPrediction)\n",
    "println(\"Accuracy: \" + accuracy)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Step 10: Validate the Logistic Regression"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "validationDataWithPrediction: org.apache.spark.sql.DataFrame = [loan_amount: double, dti: double ... 22 more fields]"
     ]
    }
   ],
   "source": [
    "// Run the feature transformers and score the validation dataset with LR model\n",
    "val validationDataWithPrediction = sparkPipelineLr.transform(validationDataset)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy: 0.9915699979510854"
     ]
    }
   ],
   "source": [
    "val evaluator = new BinaryClassificationEvaluator().\n",
    "    setLabelCol(logisticRegression.getLabelCol).\n",
    "    setRawPredictionCol(logisticRegression.getRawPredictionCol).\n",
    "    setMetricName(\"areaUnderROC\")\n",
    "    \n",
    "val accuracy = evaluator.evaluate(validationDataWithPrediction)\n",
    "println(\"Accuracy: \" + accuracy)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Part 4: Spark ML Hyperparameter Tuning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "// Define our ParamGridBuilder\n",
    "val paramGrid = new ParamGridBuilder().\n",
    "      addGrid(randomForest.maxDepth, Array(6, 8, 10)).\n",
    "      addGrid(randomForest.numTrees, Array(10, 20, 25)).\n",
    "      build()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "// Define our Validator\n",
    "val evaluator = new BinaryClassificationEvaluator().\n",
    "    setLabelCol(randomForest.getLabelCol).\n",
    "    setRawPredictionCol(randomForest.getRawPredictionCol).\n",
    "    setMetricName(\"areaUnderROC\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "// Define the CrossValidator\n",
    "val cv = new CrossValidator().\n",
    "      setEstimator(sparkPipelineEstimatorRf).\n",
    "      setEvaluator(evaluator).\n",
    "      setEstimatorParamMaps(paramGrid).\n",
    "      setNumFolds(3)\n",
    "    \n",
    "val cvPipeline = cv.fit(trainingDataset)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "val bestPipeline = cvPipeline.bestModel.asInstanceOf[PipelineModel]\n",
    "bestPipeline.stages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "val bestRf = bestPipeline.stages(1).asInstanceOf[RandomForestClassificationModel]\n",
    "bestRf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "println(\"Best Number of Trees: \" + bestRf.getNumTrees)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "println(\"Best Max Depth: \" + bestRf.getMaxDepth)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "println(bestRf.explainParams)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "val validationDataWithPrediction = bestPipeline.transform(validationDataset)\n",
    "\n",
    "val evaluator = new BinaryClassificationEvaluator().\n",
    "    setLabelCol(logisticRegression.getLabelCol).\n",
    "    setRawPredictionCol(logisticRegression.getRawPredictionCol).\n",
    "    setMetricName(\"areaUnderROC\")\n",
    "    \n",
    "val accuracy = evaluator.evaluate(validationDataWithPrediction)\n",
    "println(\"Accuracy: \" + accuracy)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Part 5: Serializing ML Pipelines for Deployment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "// Compute the bundle context, which contains the metadata that describes how we trained the pipeline\n",
    "val sbc = SparkBundleContext().withDataset(sparkPipelineLr.transform(datasetImputedFiltered))\n",
    "\n",
    "for(bf <- managed(BundleFile(\"jar:file:/tmp/lc.model.lr_demo.zip\"))) {\n",
    "        sparkPipelineLr.writeBundle.save(bf)(sbc).get\n",
    "      }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "\n",
    "val sbcRf = SparkBundleContext().withDataset(sparkPipelineRf.transform(datasetImputedFiltered))\n",
    "for(bf <- managed(BundleFile(\"jar:file:/tmp/lc.model.rf.zip\"))) {\n",
    "        sparkPipelineRf.writeBundle.save(bf)(sbcRf).get\n",
    "      }\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Part 6: Deploy Model with cURL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{}"
     ]
    }
   ],
   "source": [
    "import sys.process._\n",
    "\n",
    "// Use curl to upload our model to the Heroku app\n",
    "// Make sure to replace <App Name> with the name\n",
    "// Heroku assigned to your app\n",
    "val herokuAppName = \"frozen-sands-62833\"\n",
    "val command = Seq(\"curl\", \"-XPUT\",\n",
    "                 \"-H\", \"content-type: application/binary\",\n",
    "                 \"--data-binary\", \"@/tmp/lc.model.lr_demo.zip\",\n",
    "                 s\"http://$herokuAppName.herokuapp.com/model\")\n",
    "val output = command !!\n",
    "\n",
    "// You should see an empty set of curly braces if this command succeeds\n",
    "// \"{}\"\n",
    "println(output)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Spark",
   "language": "",
   "name": "sparkkernel"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}