decison_tree.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark= SparkSession.builder.getOrCreate()\n",
    "\n",
    "# reading in the data\n",
    "infile = 'default.csv'\n",
    "df = spark.read.csv(infile, inferSchema=True, header = True)\n",
    "\n",
    "# rename the response as \"label\"\n",
    "df = df.withColumnRenamed(\"default payment next month\", \"label\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "# balance the data \n",
    "# Split data into train set (70%), test set (30%) using seed=314\n",
    "training, test = df.randomSplit([0.7, 0.3], seed=314)\n",
    "\n",
    "# Balance the data using oversampling method. Do this on train set only. \n",
    "from pyspark.sql.functions import col, explode, array, lit\n",
    "major_train = training.filter(col('label') == 0)\n",
    "minor_train = training.filter(col('label') == 1)\n",
    "ratio = int(major_train.count()/minor_train.count())\n",
    "ratio_range = range(ratio)\n",
    "oversampled_train = minor_train.withColumn(\"dummy\", explode(array([lit(x) for x in ratio_range]))).drop('dummy')\n",
    "final_train = major_train.unionAll(oversampled_train)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Decision Tree"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Decision Tree pipeline\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.ml.classification import DecisionTreeClassifier\n",
    "from pyspark.ml import Pipeline\n",
    "\n",
    "assembler_dt = VectorAssembler(inputCols=['LIMIT_BAL','SEX',\n",
    "                       'EDUCATION','MARRIAGE','AGE',\n",
    "                       'PAY_0','PAY_2','PAY_3',\n",
    "                       'PAY_4','PAY_5','PAY_6',\n",
    "                       'BILL_AMT1','BILL_AMT2','BILL_AMT3',\n",
    "                       'BILL_AMT4','BILL_AMT5','BILL_AMT6',\n",
    "                       'PAY_AMT1','PAY_AMT2','PAY_AMT3',\n",
    "                       'PAY_AMT4','PAY_AMT5','PAY_AMT6'], outputCol=\"features\")\n",
    "\n",
    "dt = DecisionTreeClassifier(labelCol=\"label\", featuresCol=\"features\")\n",
    "pipeline_dt = Pipeline(stages=[assembler_dt,dt])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "# model selection\n",
    "\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "\n",
    "# ParamGrid for cross validation\n",
    "paramGrid_dt = ParamGridBuilder()\\\n",
    "             .addGrid(dt.maxDepth, [2, 5, 10])\\\n",
    "             .addGrid(dt.maxBins, [5, 10, 20])\\\n",
    "             .build()\n",
    "            \n",
    "# use pipeline as an estimator, 10-fold cross validation, seed 314\n",
    "crossVal_dt = CrossValidator(estimator = pipeline_dt,\n",
    "                          estimatorParamMaps = paramGrid_dt,\n",
    "                          evaluator = BinaryClassificationEvaluator(),\n",
    "                          numFolds = 10,\n",
    "                          seed = 314)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [],
   "source": [
    "# fit models\n",
    "cv_model_dt = crossVal_dt.fit(final_train)\n",
    "\n",
    "# make predictions\n",
    "prediction_dt = cv_model_dt.transform(test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Model Evaluation for Decision Tree"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.mllib.evaluation import BinaryClassificationMetrics\n",
    "from pyspark.mllib.evaluation import MulticlassMetrics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Confusion Matrix:\n",
      "[[5757. 1322.]\n",
      " [ 800. 1196.]]\n"
     ]
    }
   ],
   "source": [
    "# confusion matrix\n",
    "label_pred_dt = prediction_dt[\"label\", \"prediction\"].rdd.map(lambda p: (float(p.prediction), float(p.label)))\n",
    "matrix_dt = MulticlassMetrics(label_pred_dt)\n",
    "\n",
    "confusionMatrix = matrix_dt.confusionMatrix().toArray()\n",
    "print(\"Confusion Matrix:\\n{}\".format(confusionMatrix))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### [TN, FP]\n",
    "### [FN, TP]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "accuracy: 0.7662\n",
      "precision: 0.4750\n",
      "recall: 0.5992\n",
      "F1 score: 0.5299\n"
     ]
    }
   ],
   "source": [
    "# accuracy, precision, recall, and F1-score\n",
    "TP_dt = prediction_dt.filter('prediction = 1 AND label = prediction').count()\n",
    "TN_dt = prediction_dt.filter('prediction = 0 AND label = prediction').count()\n",
    "FP_dt = prediction_dt.filter('prediction = 1 AND label <> prediction').count()\n",
    "FN_dt = prediction_dt.filter('prediction = 0 AND label <> prediction').count()\n",
    "\n",
    "accuracy_dt = (TN_dt + TP_dt) / (TN_dt + TP_dt + FN_dt + FP_dt)\n",
    "precision_dt = TP_dt / (TP_dt + FP_dt)\n",
    "recall_dt = TP_dt / (TP_dt+ FN_dt)\n",
    "F1_dt =  2 * ((precision_dt*recall_dt) / (precision_dt + recall_dt))\n",
    "\n",
    "print(\"accuracy: %.4F\" % accuracy_dt)\n",
    "print(\"precision: %.4F\" % precision_dt)\n",
    "print(\"recall: %.4F\" % recall_dt)\n",
    "print(\"F1 score: %.4F\" % F1_dt)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Area Under ROC: 0.6801\n"
     ]
    }
   ],
   "source": [
    "# Area Under ROC\n",
    "eval_dt = BinaryClassificationEvaluator()\n",
    "auc_dt = eval_dt.evaluate(prediction_dt, {eval_dt.metricName: \"areaUnderROC\"})\n",
    "print(\"Area Under ROC: %.4F\" % auc_dt)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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": 4
}