examples/notebooks/python/mortgage-gpu.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction to XGBoost Spark with GPU\n",
    "\n",
    "Mortgage is an example of xgboost classifier. In this notebook, we will show you how to load data, train the xgboost model and use this model to predict if someone is \"delinquency\".\n",
    "\n",
    "A few libraries are required:\n",
    "  1. NumPy\n",
    "  2. cudf jar\n",
    "  2. xgboost4j jar\n",
    "  3. xgboost4j-spark jar"
   ]
  },
  {
   "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": [
    "Note on CPU version: Data reader is created with `spark.read` instead of `GpuDataReader(spark)`. Also vectorization is required, 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",
    "    '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.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
}