examples/notebooks/python/mortgage-gpu.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction to XGBoost Spark with GPU\n",
    "\n",
    "The goal of this notebook is to show how to train a XGBoost Model with Spark RAPIDS XGBoost library on GPUs. The dataset used with this notebook is derived from Fannie Mae’s Single-Family Loan Performance Data with all rights reserved by Fannie Mae. This processed dataset is redistributed with permission and consent from Fannie Mae. This notebook uses XGBoost to train 12-month mortgage loan delinquency prediction model .\n",
    "\n",
    "A few libraries required for this notebook:\n",
    "  1. NumPy\n",
    "  2. cudf jar\n",
    "  3. xgboost4j jar\n",
    "  4. xgboost4j-spark jar",
    "This notebook illustrates the ease of porting a sample CPU based Spark xgboost4j code into GPU. There are two changes required for running Spark XGBoost on GPU:\n",
    "  1. Data reader uses `GpuDataReader(spark)` instead of `spark.read`",
    "  2. `setFeaturesCol` function on CPU is replaced by setFeaturesCols(features)",
    "     a. This also eliminates the need for vectorization (assembling multiple feature columns in to one column) since we can read multiple columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Import All Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from ml.dmlc.xgboost4j.scala.spark import XGBoostClassificationModel, XGBoostClassifier\n",
    "from ml.dmlc.xgboost4j.scala.spark.rapids import GpuDataReader\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import FloatType, IntegerType, StructField, StructType\n",
    "from time import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note on CPU version: `GpuDataReader` is not necessary, but two extra libraries are required.\n",
    "```Python\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.sql.functions import col\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Specify the Data Schema and Load the Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "label = 'delinquency_12'\n",
    "schema = StructType([\n",
    "    StructField('orig_channel', FloatType()),\n",
    "    StructField('first_home_buyer', FloatType()),\n",
    "    StructField('loan_purpose', FloatType()),\n",
    "    StructField('property_type', FloatType()),\n",
    "    StructField('occupancy_status', FloatType()),\n",
    "    StructField('property_state', FloatType()),\n",
    "    StructField('product_type', FloatType()),\n",
    "    StructField('relocation_mortgage_indicator', FloatType()),\n",
    "    StructField('seller_name', FloatType()),\n",
    "    StructField('mod_flag', FloatType()),\n",
    "    StructField('orig_interest_rate', FloatType()),\n",
    "    StructField('orig_upb', IntegerType()),\n",
    "    StructField('orig_loan_term', IntegerType()),\n",
    "    StructField('orig_ltv', FloatType()),\n",
    "    StructField('orig_cltv', FloatType()),\n",
    "    StructField('num_borrowers', FloatType()),\n",
    "    StructField('dti', FloatType()),\n",
    "    StructField('borrower_credit_score', FloatType()),\n",
    "    StructField('num_units', IntegerType()),\n",
    "    StructField('zip', IntegerType()),\n",
    "    StructField('mortgage_insurance_percent', FloatType()),\n",
    "    StructField('current_loan_delinquency_status', IntegerType()),\n",
    "    StructField('current_actual_upb', FloatType()),\n",
    "    StructField('interest_rate', FloatType()),\n",
    "    StructField('loan_age', FloatType()),\n",
    "    StructField('msa', FloatType()),\n",
    "    StructField('non_interest_bearing_upb', FloatType()),\n",
    "    StructField(label, IntegerType()),\n",
    "])\n",
    "features = [ x.name for x in schema if x.name != label ]\n",
    "\n",
    "train_data = GpuDataReader(spark).schema(schema).option('header', True).csv('/data/datasets/mortgage-small/train')\n",
    "eval_data = GpuDataReader(spark).schema(schema).option('header', True).csv('/data/datasets/mortgage-small/eval')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "CPU version: CPU based Spark needs vectorization, which means you need to assemble all feature columns into one column.\n",
    "```Python\n",
    "def vectorize(data_frame):\n",
    "    to_floats = [ col(x.name).cast(FloatType()) for x in data_frame.schema ]\n",
    "    return (VectorAssembler()\n",
    "        .setInputCols(features)\n",
    "        .setOutputCol('features')\n",
    "        .transform(data_frame.select(to_floats))\n",
    "        .select(col('features'), col(label)))\n",
    "\n",
    "train_data = spark.read.schema(schema).option('header', True).csv('/data/datasets/mortgage-small/train')\n",
    "eval_data = spark.read.schema(schema).option('header', True).csv('/data/datasets/mortgage-small/eval')\n",
    "\n",
    "train_data = vectorize(train_data)\n",
    "eval_data = vectorize(eval_data)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create XGBoostClassifier"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "params = { \n",
    "    'eta': 0.1,\n",
    "    'gamma': 0.1,\n",
    "    'missing': 0.0,\n",
    "    'treeMethod': 'gpu_hist',\n",
    "    'maxDepth': 10, \n",
    "    'maxLeaves': 256,\n",
    "    'growPolicy': 'depthwise',\n",
    "    'objective': 'binary:logistic',\n",
    "    'minChildWeight': 30.0,\n",
    "    'lambda_': 1.0,\n",
    "    'scalePosWeight': 2.0,\n",
    "    'subsample': 1.0,\n",
    "    'nthread': 1,\n",
    "    'numRound': 100,\n",
    "    'numWorkers': 1,\n",
    "}\n",
    "classifier = XGBoostClassifier(**params).setLabelCol(label).setFeaturesCols(features)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note on CPU version: The CPU version provides the `setFeaturesCol` function, that's why vectorization is required. The parameter `num_workers` should be set to the number of machines with GPU in Spark cluster in GPU version, while it can be set to the number of your CPU cores in CPU version. The tree method `gpu_hist` is designed for GPU training, while tree method `hist` is designed for CPU training.\n",
    "```Python\n",
    "classifier = XGBoostClassifier(**params).setLabelCol(label).setFeaturesCol('features')\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Train the Data with Benchmark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Training takes 6.48 seconds\n"
     ]
    }
   ],
   "source": [
    "def with_benchmark(phrase, action):\n",
    "    start = time()\n",
    "    result = action()\n",
    "    end = time()\n",
    "    print('{} takes {} seconds'.format(phrase, round(end - start, 2)))\n",
    "    return result\n",
    "model = with_benchmark('Training', lambda: classifier.fit(train_data))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Save and Reload the Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.write().overwrite().save('/data/new-model-path')\n",
    "loaded_model = XGBoostClassificationModel().load('/data/new-model-path')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Transformation and Show Result Sample"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Transformation takes 3.2 seconds\n",
      "+--------------+--------------------+--------------------+----------+\n",
      "|delinquency_12|       rawPrediction|         probability|prediction|\n",
      "+--------------+--------------------+--------------------+----------+\n",
      "|           0.0|[0.00250339508056...|[1.00250339508056...|       0.0|\n",
      "|           0.0|[0.00229310989379...|[1.00229310989379...|       0.0|\n",
      "|           0.0|[0.00229310989379...|[1.00229310989379...|       0.0|\n",
      "|           0.0|[-0.0013450086116...|[0.99865499138832...|       0.0|\n",
      "|           0.0|[-1.2058019638061...|[0.99987941980361...|       0.0|\n",
      "+--------------+--------------------+--------------------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "def transform():\n",
    "    result = loaded_model.transform(eval_data).cache()\n",
    "    result.foreachPartition(lambda _: None)\n",
    "    return result\n",
    "result = with_benchmark('Transformation', transform)\n",
    "result.select(label, 'rawPrediction', 'probability', 'prediction').show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Evaluation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Evaluation takes 0.68 seconds\n",
      "Accuracy is 0.9980699597729774\n"
     ]
    }
   ],
   "source": [
    "accuracy = with_benchmark(\n",
    "    'Evaluation',\n",
    "    lambda: MulticlassClassificationEvaluator().setLabelCol(label).evaluate(result))\n",
    "print('Accuracy is ' + str(accuracy))"
   ]
  }
 ],
 "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}