4-xgboost_training.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "de1bbef4-6cb2-4229-9b9e-ca4c39c2e157",
   "metadata": {},
   "source": [
    "# XGBOOST Training\n",
    "The full data set is expected to exceed the memory of a single GPU so a distributed XGBOOST training strategy is applied. In situations where data fits nicely on a single machine, avoiding distributed training is recommended for improved performance. Prototyping the ML model involves running a few years of data through “3-etl_test_train_split_tgt_enc.ipynb” to generate the fully encoded wide-form tables for the various split datasets. \n",
    "\n",
    "We would like to predict DELAY_CAUSES_ENC, which contains multiple delay causes (LATE_AIRCRAFT_DELAY, CARRIER_DELAY, NAS_DELAY, WEATHER_DELAY, SECURITY_DELAY). SECURITY_DELAY and WEATHER_DELAY occur infrequently so they are combined into OTHER_DELAY. The first letter of each delay cause is used in a binary sequence (e.g., LCNO) in the DELAY_CAUSES_ENC column. The overall result is 4 unique delay causes with 16 possible combinations. To further reduce the number of meta-classes, a 95th percentile cutoff filter is applied to the ranked frequency of each meta-class. Infrequently occurring classes all end with “O” (for OTHER) in their DELAY_CAUSES_ENC and are encoded with the number 8 in the final DELAY_CAUSES_ENC representation. Including the no-delay case (“----”, or 0), there are a total of 9 meta-classes to consider.  \n",
    "\n",
    "XGBOOST’s default treatment of multi-class classification problems involve decomposing the problem using One-vs-Rest (OVR). This decomposition strategy treats the multi-class classification problem as N-binary classification sub-problems, with N being the number of distinct categories to be predicted (N=9 in our case). Reducing the number of classes while still being able to faithfully represent the data is desirable. Because N different binary classification models are created, inferencing and model explainability becomes much more computationally intensive. New data needs to pass through each binary classification sub-model and then blended into a final prediction.    \n",
    "\n",
    "Challenges: (1) severe class imbalance; (2) large amounts of class overlap between certain classes. Class imbalance can be addressed via applying class weights during xgboost training or implementing resampling techniques. Combining related classes may help with the class overlap problem.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "04799d25",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "xgboost version 1.5.0\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/btong/miniconda3/envs/rapids-21.12/lib/python3.8/site-packages/distributed/node.py:160: UserWarning: Port 8787 is already in use.\n",
      "Perhaps you already have a cluster running?\n",
      "Hosting the HTTP server on port 34381 instead\n",
      "  warnings.warn(\n",
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n",
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "    <div style=\"width: 24px; height: 24px; background-color: #e1e1e1; border: 3px solid #9D9D9D; border-radius: 5px; position: absolute;\"> </div>\n",
       "    <div style=\"margin-left: 48px;\">\n",
       "        <h3 style=\"margin-bottom: 0px;\">Client</h3>\n",
       "        <p style=\"color: #9D9D9D; margin-bottom: 0px;\">Client-2b8fb128-806f-11ec-8e9c-a87eeae84103</p>\n",
       "        <table style=\"width: 100%; text-align: left;\">\n",
       "\n",
       "        <tr>\n",
       "        \n",
       "            <td style=\"text-align: left;\"><strong>Connection method:</strong> Cluster object</td>\n",
       "            <td style=\"text-align: left;\"><strong>Cluster type:</strong> dask_cuda.LocalCUDACluster</td>\n",
       "        \n",
       "        </tr>\n",
       "\n",
       "        \n",
       "            <tr>\n",
       "                <td style=\"text-align: left;\">\n",
       "                    <strong>Dashboard: </strong> <a href=\"http://127.0.0.1:34381/status\" target=\"_blank\">http://127.0.0.1:34381/status</a>\n",
       "                </td>\n",
       "                <td style=\"text-align: left;\"></td>\n",
       "            </tr>\n",
       "        \n",
       "\n",
       "        </table>\n",
       "\n",
       "        \n",
       "            <details>\n",
       "            <summary style=\"margin-bottom: 20px;\"><h3 style=\"display: inline;\">Cluster Info</h3></summary>\n",
       "            <div class=\"jp-RenderedHTMLCommon jp-RenderedHTML jp-mod-trusted jp-OutputArea-output\">\n",
       "    <div style=\"width: 24px; height: 24px; background-color: #e1e1e1; border: 3px solid #9D9D9D; border-radius: 5px; position: absolute;\">\n",
       "    </div>\n",
       "    <div style=\"margin-left: 48px;\">\n",
       "        <h3 style=\"margin-bottom: 0px; margin-top: 0px;\">LocalCUDACluster</h3>\n",
       "        <p style=\"color: #9D9D9D; margin-bottom: 0px;\">856c0f96</p>\n",
       "        <table style=\"width: 100%; text-align: left;\">\n",
       "            <tr>\n",
       "                <td style=\"text-align: left;\">\n",
       "                    <strong>Dashboard:</strong> <a href=\"http://127.0.0.1:34381/status\" target=\"_blank\">http://127.0.0.1:34381/status</a>\n",
       "                </td>\n",
       "                <td style=\"text-align: left;\">\n",
       "                    <strong>Workers:</strong> 2\n",
       "                </td>\n",
       "            </tr>\n",
       "            <tr>\n",
       "                <td style=\"text-align: left;\">\n",
       "                    <strong>Total threads:</strong> 32\n",
       "                </td>\n",
       "                <td style=\"text-align: left;\">\n",
       "                    <strong>Total memory:</strong> 238.42 GiB\n",
       "                </td>\n",
       "            </tr>\n",
       "            \n",
       "            <tr>\n",
       "    <td style=\"text-align: left;\"><strong>Status:</strong> running</td>\n",
       "    <td style=\"text-align: left;\"><strong>Using processes:</strong> True</td>\n",
       "</tr>\n",
       "\n",
       "            \n",
       "        </table>\n",
       "\n",
       "        <details>\n",
       "            <summary style=\"margin-bottom: 20px;\">\n",
       "                <h3 style=\"display: inline;\">Scheduler Info</h3>\n",
       "            </summary>\n",
       "\n",
       "            <div style=\"\">\n",
       "    <div>\n",
       "        <div style=\"width: 24px; height: 24px; background-color: #FFF7E5; border: 3px solid #FF6132; border-radius: 5px; position: absolute;\"> </div>\n",
       "        <div style=\"margin-left: 48px;\">\n",
       "            <h3 style=\"margin-bottom: 0px;\">Scheduler</h3>\n",
       "            <p style=\"color: #9D9D9D; margin-bottom: 0px;\">Scheduler-363dc876-5d61-400a-9c1a-3733185139ea</p>\n",
       "            <table style=\"width: 100%; text-align: left;\">\n",
       "                <tr>\n",
       "                    <td style=\"text-align: left;\">\n",
       "                        <strong>Comm:</strong> tcp://127.0.0.1:41505\n",
       "                    </td>\n",
       "                    <td style=\"text-align: left;\">\n",
       "                        <strong>Workers:</strong> 2\n",
       "                    </td>\n",
       "                </tr>\n",
       "                <tr>\n",
       "                    <td style=\"text-align: left;\">\n",
       "                        <strong>Dashboard:</strong> <a href=\"http://127.0.0.1:34381/status\" target=\"_blank\">http://127.0.0.1:34381/status</a>\n",
       "                    </td>\n",
       "                    <td style=\"text-align: left;\">\n",
       "                        <strong>Total threads:</strong> 32\n",
       "                    </td>\n",
       "                </tr>\n",
       "                <tr>\n",
       "                    <td style=\"text-align: left;\">\n",
       "                        <strong>Started:</strong> Just now\n",
       "                    </td>\n",
       "                    <td style=\"text-align: left;\">\n",
       "                        <strong>Total memory:</strong> 238.42 GiB\n",
       "                    </td>\n",
       "                </tr>\n",
       "            </table>\n",
       "        </div>\n",
       "    </div>\n",
       "\n",
       "    <details style=\"margin-left: 48px;\">\n",
       "        <summary style=\"margin-bottom: 20px;\">\n",
       "            <h3 style=\"display: inline;\">Workers</h3>\n",
       "        </summary>\n",
       "\n",
       "        \n",
       "        <div style=\"margin-bottom: 20px;\">\n",
       "            <div style=\"width: 24px; height: 24px; background-color: #DBF5FF; border: 3px solid #4CC9FF; border-radius: 5px; position: absolute;\"> </div>\n",
       "            <div style=\"margin-left: 48px;\">\n",
       "            <details>\n",
       "                <summary>\n",
       "                    <h4 style=\"margin-bottom: 0px; display: inline;\">Worker: 0</h4>\n",
       "                </summary>\n",
       "                <table style=\"width: 100%; text-align: left;\">\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Comm: </strong> tcp://127.0.0.1:38607\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Total threads: </strong> 16\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Dashboard: </strong> <a href=\"http://127.0.0.1:41145/status\" target=\"_blank\">http://127.0.0.1:41145/status</a>\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Memory: </strong> 119.21 GiB\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Nanny: </strong> tcp://127.0.0.1:44877\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\"></td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td colspan=\"2\" style=\"text-align: left;\">\n",
       "                            <strong>Local directory: </strong> /home/btong/github/flight-delay-causes/dask-worker-space/worker-vrqpca4x\n",
       "                        </td>\n",
       "                    </tr>\n",
       "\n",
       "                    \n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>GPU: </strong>Quadro RTX 8000\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>GPU memory: </strong> 48.00 GiB\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    \n",
       "\n",
       "                    \n",
       "\n",
       "                </table>\n",
       "            </details>\n",
       "            </div>\n",
       "        </div>\n",
       "        \n",
       "        <div style=\"margin-bottom: 20px;\">\n",
       "            <div style=\"width: 24px; height: 24px; background-color: #DBF5FF; border: 3px solid #4CC9FF; border-radius: 5px; position: absolute;\"> </div>\n",
       "            <div style=\"margin-left: 48px;\">\n",
       "            <details>\n",
       "                <summary>\n",
       "                    <h4 style=\"margin-bottom: 0px; display: inline;\">Worker: 1</h4>\n",
       "                </summary>\n",
       "                <table style=\"width: 100%; text-align: left;\">\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Comm: </strong> tcp://127.0.0.1:39591\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Total threads: </strong> 16\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Dashboard: </strong> <a href=\"http://127.0.0.1:46003/status\" target=\"_blank\">http://127.0.0.1:46003/status</a>\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Memory: </strong> 119.21 GiB\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Nanny: </strong> tcp://127.0.0.1:34905\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\"></td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td colspan=\"2\" style=\"text-align: left;\">\n",
       "                            <strong>Local directory: </strong> /home/btong/github/flight-delay-causes/dask-worker-space/worker-dpeue4ow\n",
       "                        </td>\n",
       "                    </tr>\n",
       "\n",
       "                    \n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>GPU: </strong>Quadro RTX 8000\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>GPU memory: </strong> 48.00 GiB\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    \n",
       "\n",
       "                    \n",
       "\n",
       "                </table>\n",
       "            </details>\n",
       "            </div>\n",
       "        </div>\n",
       "        \n",
       "\n",
       "    </details>\n",
       "</div>\n",
       "\n",
       "        </details>\n",
       "    </div>\n",
       "</div>\n",
       "            </details>\n",
       "        \n",
       "\n",
       "    </div>\n",
       "</div>"
      ],
      "text/plain": [
       "<Client: 'tcp://127.0.0.1:41505' processes=2 threads=32, memory=238.42 GiB>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import gc\n",
    "import os\n",
    "import sys\n",
    "import glob\n",
    "import shutil\n",
    "import pandas as pd\n",
    "import pyarrow as pa\n",
    "import pyarrow.parquet as pq\n",
    "from pandas.api.types import CategoricalDtype\n",
    "import numpy as np\n",
    "from time import time\n",
    "from datetime import datetime\n",
    "\n",
    "import dask\n",
    "import dask.dataframe as dd\n",
    "from dask.distributed import Client, wait, progress, get_worker\n",
    "\n",
    "import sklearn\n",
    "from sklearn.model_selection import train_test_split\n",
    "import xgboost as xgb\n",
    "from xgboost.dask import DaskDMatrix\n",
    "\n",
    "print('xgboost version', xgb.__version__)\n",
    "\n",
    "run_type = 'gpu'\n",
    "max_depth = 15\n",
    "num_boost_round = 200\n",
    "score_metric = 'auc'\n",
    "storage_backend = 'local'\n",
    "\n",
    "test_size = 0.01 # Non-zero to pass data split checker. Value not used since data previously partitioned into test/train folders.\n",
    "   \n",
    "study_arpt = 'NAS'\n",
    "\n",
    "if study_arpt == 'NAS':\n",
    "    # NAS processing excludes weather. Has additional cols for arrival/departure airports.\n",
    "    pred_model = 'multi_class'\n",
    "    label_col = 'DELAY_CAUSES_ENC'\n",
    "    excluded_features = [label_col, 'cv_idx', 'class_weight', 'UID', 'ARR_DEL15', 'DEP_DEL15', 'YYYYMM']\n",
    "else:\n",
    "    pred_model = 'binary_class'\n",
    "    label_col = 'ARR_DEL15'\n",
    "    excluded_features = [label_col, 'cv_idx', 'UID', 'DEL_ARR_PER_QTHR', 'DEL_DEP_PER_QTHR', 'DEP_DEL15', 'ARR_DEL']\n",
    "\n",
    "if pred_model == 'binary_class':\n",
    "    xgb_objective = 'binary:logistic'\n",
    "elif pred_model == 'multi_class':\n",
    "    # xgboost auc docs mentioned that: \"When used with multi-class classification, objective should be multi:softprob instead of multi:softmax, \n",
    "    # as the latter doesn’t output probability. Also the AUC is calculated by 1-vs-rest with reference class weighted by class prevalence.\"\n",
    "    xgb_objective = 'multi:softprob'\n",
    "# elif pred_model == 'regression':\n",
    "#     label_col = 'ARR_DELAY' # Regression model\n",
    "#     xgb_objective = 'reg:squarederror'\n",
    "    \n",
    "enc_output_dir = './data/encoded/split/'+study_arpt\n",
    "\n",
    "xgb_model_name = 'xgb_'+run_type+'_airline_delay_'+study_arpt+'_max_depth_'+str(max_depth)\n",
    "\n",
    "\n",
    "if run_type == 'gpu':\n",
    "    from dask_cuda import LocalCUDACluster\n",
    "    \n",
    "    # Run on all available GPU on same machine:\n",
    "    cluster = LocalCUDACluster(threads_per_worker=16, memory_limit='128GB')\n",
    "    client = Client(cluster)\n",
    "    \n",
    "    # Run single GPU:\n",
    "#     client = Client(n_workers=1, threads_per_worker=32)\n",
    "elif run_type == 'cpu':\n",
    "    client = Client(n_workers=1, threads_per_worker=32)\n",
    "#     client = Client(n_workers=2, threads_per_worker=16)   \n",
    "    \n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "793e40bb",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of classes detected:  5\n",
      "Starting xgboost gpu multi_class training....\n",
      "Feature columns: ['YEAR', 'DEP_DELAY', 'ARR_DELAY', 'TAXI_OUT', 'TAXI_IN', 'CRS_ELAPSED_TIME', 'ACTUAL_ELAPSED_TIME', 'AIR_TIME', 'DISTANCE', 'DISTANCE_GROUP', 'DEP_ARPT_LAT', 'DEP_ARPT_LON', 'DEP_ARPT_ELV_FT', 'ARR_ARPT_LAT', 'ARR_ARPT_LON', 'ARR_ARPT_ELV_FT', 'IS_HOLIDAY', 'IS_WEEKEND', 'DEP_ARPT_TYPE_large_airport', 'DEP_ARPT_TYPE_medium_airport', 'DEP_ARPT_TYPE_small_airport', 'ARR_ARPT_TYPE_large_airport', 'ARR_ARPT_TYPE_medium_airport', 'ARR_ARPT_TYPE_small_airport', 'QUARTER_cos', 'QUARTER_sin', 'MONTH_cos', 'MONTH_sin', 'DAY_OF_MONTH_cos', 'DAY_OF_MONTH_sin', 'DAY_OF_YEAR_cos', 'DAY_OF_YEAR_sin', 'DAY_OF_WEEK_cos', 'DAY_OF_WEEK_sin', 'CRS_DEP_TIME_HR_cos', 'CRS_DEP_TIME_HR_sin', 'CRS_DEP_TIME_QTHR_cos', 'CRS_DEP_TIME_QTHR_sin', 'CRS_ARR_TIME_HR_cos', 'CRS_ARR_TIME_HR_sin', 'CRS_ARR_TIME_QTHR_cos', 'CRS_ARR_TIME_QTHR_sin', 'ORIGIN_ARR_PER_QTHR', 'ORIGIN_CRS_ARR_PER_QTHR', 'ORIGIN_DEL_ARR_PER_QTHR', 'ORIGIN_ARR_DIFF0_PER_QTHR', 'ORIGIN_ARR_PER_QTHR_LAG30', 'ORIGIN_CRS_ARR_PER_QTHR_LAG30', 'ORIGIN_DEL_ARR_PER_QTHR_LAG30', 'ORIGIN_ARR_DIFF0_PER_QTHR_LAG30', 'ORIGIN_ARR_PER_QTHR_LAG15', 'ORIGIN_CRS_ARR_PER_QTHR_LAG15', 'ORIGIN_DEL_ARR_PER_QTHR_LAG15', 'ORIGIN_ARR_DIFF0_PER_QTHR_LAG15', 'ORIGIN_CRS_ARR_PER_QTHR_LEAD15', 'ORIGIN_CRS_ARR_PER_QTHR_LEAD30', 'ORIGIN_ARR_DIFF1_PER_QTHR', 'ORIGIN_ARR_DIFF2_PER_QTHR', 'ORIGIN_DEP_PER_QTHR', 'ORIGIN_CRS_DEP_PER_QTHR', 'ORIGIN_DEL_DEP_PER_QTHR', 'ORIGIN_DEP_DIFF0_PER_QTHR', 'ORIGIN_DEP_PER_QTHR_LAG30', 'ORIGIN_CRS_DEP_PER_QTHR_LAG30', 'ORIGIN_DEL_DEP_PER_QTHR_LAG30', 'ORIGIN_DEP_DIFF0_PER_QTHR_LAG30', 'ORIGIN_DEP_PER_QTHR_LAG15', 'ORIGIN_CRS_DEP_PER_QTHR_LAG15', 'ORIGIN_DEL_DEP_PER_QTHR_LAG15', 'ORIGIN_DEP_DIFF0_PER_QTHR_LAG15', 'ORIGIN_CRS_DEP_PER_QTHR_LEAD15', 'ORIGIN_CRS_DEP_PER_QTHR_LEAD30', 'ORIGIN_DEP_DIFF1_PER_QTHR', 'ORIGIN_DEP_DIFF2_PER_QTHR', 'DEST_ARR_PER_QTHR', 'DEST_CRS_ARR_PER_QTHR', 'DEST_DEL_ARR_PER_QTHR', 'DEST_ARR_DIFF0_PER_QTHR', 'DEST_ARR_PER_QTHR_LAG30', 'DEST_CRS_ARR_PER_QTHR_LAG30', 'DEST_DEL_ARR_PER_QTHR_LAG30', 'DEST_ARR_DIFF0_PER_QTHR_LAG30', 'DEST_ARR_PER_QTHR_LAG15', 'DEST_CRS_ARR_PER_QTHR_LAG15', 'DEST_DEL_ARR_PER_QTHR_LAG15', 'DEST_ARR_DIFF0_PER_QTHR_LAG15', 'DEST_CRS_ARR_PER_QTHR_LEAD15', 'DEST_CRS_ARR_PER_QTHR_LEAD30', 'DEST_ARR_DIFF1_PER_QTHR', 'DEST_ARR_DIFF2_PER_QTHR', 'DEST_DEP_PER_QTHR', 'DEST_CRS_DEP_PER_QTHR', 'DEST_DEL_DEP_PER_QTHR', 'DEST_DEP_DIFF0_PER_QTHR', 'DEST_DEP_PER_QTHR_LAG30', 'DEST_CRS_DEP_PER_QTHR_LAG30', 'DEST_DEL_DEP_PER_QTHR_LAG30', 'DEST_DEP_DIFF0_PER_QTHR_LAG30', 'DEST_DEP_PER_QTHR_LAG15', 'DEST_CRS_DEP_PER_QTHR_LAG15', 'DEST_DEL_DEP_PER_QTHR_LAG15', 'DEST_DEP_DIFF0_PER_QTHR_LAG15', 'DEST_CRS_DEP_PER_QTHR_LEAD15', 'DEST_CRS_DEP_PER_QTHR_LEAD30', 'DEST_DEP_DIFF1_PER_QTHR', 'DEST_DEP_DIFF2_PER_QTHR', 'ORIGIN_HourlyDryBulbTemperature', 'ORIGIN_HourlyPrecipitation', 'ORIGIN_HourlyPressureChange', 'ORIGIN_HourlyVisibility', 'ORIGIN_HourlyWindGustSpeed', 'ORIGIN_HourlyWindSpeed', 'ORIGIN_HourlySkyCover_BKN', 'ORIGIN_HourlySkyCover_FEW', 'ORIGIN_HourlySkyCover_OVC', 'ORIGIN_HourlySkyCover_SCT', 'ORIGIN_HourlySkyCover_VV', 'ORIGIN_HourlyWindDirection_cos', 'ORIGIN_HourlyWindDirection_sin', 'DEST_HourlyDryBulbTemperature', 'DEST_HourlyPrecipitation', 'DEST_HourlyPressureChange', 'DEST_HourlyVisibility', 'DEST_HourlyWindGustSpeed', 'DEST_HourlyWindSpeed', 'DEST_HourlySkyCover_BKN', 'DEST_HourlySkyCover_FEW', 'DEST_HourlySkyCover_OVC', 'DEST_HourlySkyCover_SCT', 'DEST_HourlySkyCover_VV', 'DEST_HourlyWindDirection_cos', 'DEST_HourlyWindDirection_sin', 'OP_UNIQUE_CARRIER', 'ORIGIN', 'DEST', 'OD_PAIR', 'HOLIDAY_NAME', 'TAIL_NUM', 'ORIGIN_HourlyPresentWeatherTypeCombo', 'DEST_HourlyPresentWeatherTypeCombo']\n",
      "  Number of feature columns: 140\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[19:19:16] task [xgboost.dask]:tcp://127.0.0.1:39591 got new rank 0\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'data_load_time': 1.21, 'train_model_time': 25.69, 'export_model_time': 0.33, 'total_time': 32.4}\n",
      "Dask xgboost gpu training time: 32.40s\n"
     ]
    }
   ],
   "source": [
    "def dask_xgb_train_from_enc(client, run_type, data_path, label_col, xgb_params, xgb_model_name):\n",
    "    \"\"\"\n",
    "    Use dask + xgboost for training against already encoded data.\n",
    "    \"\"\"\n",
    "    if run_type == 'cpu':\n",
    "        ddf_enc = dd.read_parquet(data_path)\n",
    "    elif run_type == 'gpu':\n",
    "        import dask_cudf\n",
    "        import cupy as cp\n",
    "        ddf_enc = dask_cudf.read_parquet(data_path)\n",
    "    else:\n",
    "        raise ValueError('Select run_type of cpu or gpu.')\n",
    "\n",
    "    tic = time()\n",
    "    ddf_enc = client.persist(ddf_enc)\n",
    "#     wait([ddf_enc])\n",
    "\n",
    "    feature_cols = [cc for cc in ddf_enc.columns if cc not in excluded_features]\n",
    "    print('Feature columns:', feature_cols)\n",
    "    print('  Number of feature columns:', len(feature_cols))\n",
    "    \n",
    "    X = ddf_enc[feature_cols].astype('float32')\n",
    "    y = ddf_enc[label_col].astype('float32')\n",
    "    \n",
    "    # Per-instance weight required due to severe class imbalance in One-vs-Rest decomposition.\n",
    "    # Weights precalculated during ETL on training set alone to avoid leakage.\n",
    "    weight = ddf_enc['class_weight'].astype('float32')\n",
    "    \n",
    "#     # Persist data:\n",
    "#     X = client.persist(X)\n",
    "#     y = client.persist(y)\n",
    "#     weight = client.persist(weight)\n",
    "#     wait([X, y, weight])\n",
    "\n",
    "    # Use XGBOOST API:\n",
    "    if run_type == 'gpu':\n",
    "        # See example: https://github.com/dmlc/xgboost/blob/master/demo/dask/gpu_training.py\n",
    "        # `DaskDeviceQuantileDMatrix` is used instead of `DaskDMatrix`, be careful\n",
    "        # that it can not be used for anything else than training.\n",
    "        dtrain = xgb.dask.DaskDeviceQuantileDMatrix(client, X, y, weight=weight)\n",
    "    else:\n",
    "        dtrain = DaskDMatrix(client, X, y, weight=weight)\n",
    "    \n",
    "#     del ddf_enc, X, y # Cleanup\n",
    "    data_load_time = np.round(time() - tic, 2)\n",
    "\n",
    "    tic = time()\n",
    "    xgb_model = xgb.dask.train(client, xgb_params, dtrain, num_boost_round=num_boost_round)['booster']\n",
    "    train_model_time = np.round(time() - tic, 2)\n",
    "\n",
    "    # Save xgb model to disk:\n",
    "    tic = time()\n",
    "    xgb_model.save_model(xgb_model_name+'.model') # Save model to current dir\n",
    "    xgb_model.save_model('/tmp/'+xgb_model_name+'.json') # Experimental JSON format. To be nested in output file with timings/metrics.\n",
    "\n",
    "    # Move local xgb-model to hdfs if xgb_hdfs_output_dir specified:\n",
    "    if (storage_backend == 'hdfs'):\n",
    "        if xgb_hdfs_output_dir != None:\n",
    "            with open(xgb_model_name+'.model','rb') as f:\n",
    "                hdfs.upload(os.path.join(xgb_hdfs_output_dir, xgb_model_name+'.model'), f)\n",
    "    export_model_time = np.round(time() - tic, 2)\n",
    "\n",
    "    metrics = {\n",
    "        'data_load_time': data_load_time,\n",
    "        'train_model_time': train_model_time,\n",
    "        'export_model_time': export_model_time\n",
    "        }\n",
    "\n",
    "    del dtrain\n",
    "    gc.collect() # Manual garbage collection needed since python may not clean up GPU resources automatically.\n",
    "    return(metrics)\n",
    "\n",
    "\n",
    "# xgboost params:\n",
    "xgb_params = { \n",
    "    'random_state': 0,\n",
    "    'eta': 0.1,\n",
    "    'gamma': 0.1,\n",
    "    'max_depth': max_depth, \n",
    "#     'num_boost_round': num_boost_round, # Number of boosting rounds: num_boost_round in xgboost, numRound in xgboost4j, n_estimators in scikit-learn.\n",
    "    'max_leaves': 4*256, # Maximum number of nodes to be added. Only relevant when grow_policy=lossguide is set.\n",
    "    'objective': xgb_objective,\n",
    "    'eval_metric': 'auc', # Applied to eval_set/test_data if provided\n",
    "    'booster': 'gbtree',\n",
    "}\n",
    "\n",
    "if run_type == 'cpu':\n",
    "    tree_method = 'hist'\n",
    "elif run_type == 'gpu':   \n",
    "    tree_method = 'gpu_hist'\n",
    "\n",
    "xgb_params['tree_method'] = tree_method\n",
    "\n",
    "if study_arpt == 'NAS':\n",
    "    y_tmp = dd.read_parquet(enc_output_dir+'/train', columns=label_col)\n",
    "    \n",
    "    # Assume that classes have been sequentially encoded:\n",
    "    num_class = y_tmp.max().compute() + 1\n",
    "    print('Number of classes detected: ', num_class)\n",
    "    xgb_params['num_class'] = num_class\n",
    "\n",
    "print('Starting xgboost '+run_type+' '+pred_model+' training....')\n",
    "\n",
    "tic = time()\n",
    "\n",
    "# Large amount of memory required to train on full 16 yrs of data. Reduce number of years used during training on single node tests.\n",
    "training_metrics = dask_xgb_train_from_enc(client, run_type, enc_output_dir+'/train', label_col, xgb_params, xgb_model_name)\n",
    "\n",
    "t_dask_train = np.round(time() - tic, 2)\n",
    "\n",
    "training_metrics['total_time'] = t_dask_train\n",
    "print(training_metrics)\n",
    "print('Dask xgboost '+run_type+' training time:', '{:0.2f}'.format(t_dask_train) + 's')\n",
    "\n",
    "# Running multi-class is similar in timing to running N_class-times binary classification due to One-vs-Rest problem decomposition."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "08830bc0-0088-4d72-8e6c-368ee24f6df4",
   "metadata": {},
   "source": [
    "Multi-class xgboost classification model with 9 meta-classes using One-vs-All class decomposition. Sixteen years of BTS airline delay data in training set, one year in test set (~111M records, 140 attributes, over ~17 year period). Data takes up ~12GB VRAM during training using xgb.dask.DaskDeviceQuantileDMatrix.  \n",
    "XGBOOST settings: max_depth=variable, num_boost_round=200.\n",
    "\n",
    "AUC scoring is performed on CPU only since cuML scoring has not implemented multi-class scorer for OVA.\n",
    "Due to model complexity, GPU is expected to perform better at inference.\n",
    "\n",
    "5-yr NAS data with 140 attr (includes weather and schedule) 201406-201903 train, 201904-202003 test. Uses ~32GB VRAM during GPU training or ~70GB during CPU training.\n",
    "\n",
    "10-yr NAS data with 140 attr (includes weather and schedule) 200906-201903 train, 201904-202003 test. Uses ~65GB VRAM during GPU training or 140GB during CPU training."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "e89834c7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting gpu multi_class inference rounds....\n",
      "Scoring predictions in ./data/encoded/split/NAS/train\n",
      "Scoring predictions in ./data/encoded/split/NAS/test\n",
      "Inference against training data:  {'data_load_time': 0.07, 'predict_time': 3.17, 'score': {'auc': 1.0, 'auc_time': 0.14}}\n",
      "Inference against (hold-out) test data:  {'data_load_time': 0.08, 'predict_time': 1.62, 'score': {'auc': 0.9938896562385731, 'auc_time': 0.17}}\n",
      "Total prediction time: 5.42s\n"
     ]
    }
   ],
   "source": [
    "def score_pred(run_type, y_true, y_pred, score_metric):\n",
    "    \"\"\"\n",
    "    Prediction scoring function.\n",
    "    \"\"\"\n",
    "    if pred_model == 'multi_class':\n",
    "        # Scoring for multi-class is done on CPU only. cuML roc_auc_score doesn't not work with multi-class yet.\n",
    "        # Work-around is to use CPU for scoring via converting gpu-dataframe to pandas.\n",
    "        if run_type == 'gpu':\n",
    "            import cupy\n",
    "            y_true = cupy.asnumpy(y_true)\n",
    "            y_pred = cupy.asnumpy(y_pred)\n",
    "            \n",
    "            # Update run_type to bypass GPU methods.\n",
    "            run_type = 'cpu'\n",
    "    \n",
    "    if run_type == 'gpu':\n",
    "#         import dask.dataframe as hw # Has internal switching for compatibility. Takes a long time to perform AUC score, but fast at prediction.\n",
    "        from cuml.metrics import roc_auc_score\n",
    "        from cuml.metrics.accuracy import accuracy_score\n",
    "        # TODO: inplace scoring to pair with inplace prediction? Only works with certain methods.\n",
    "    else:\n",
    "        # TODO: need parallelized roc_auc_score computation for auc.\n",
    "        from sklearn.metrics import roc_auc_score # Single-threaded?\n",
    "#         from sklearn.metrics import accuracy_score # Single-threaded\n",
    "        from dask_ml.metrics import accuracy_score # CPU parallelized\n",
    "\n",
    "    if score_metric == 'auc':\n",
    "        if pred_model == 'multi_class':\n",
    "            # xgb multi-class classification uses One-vs-Rest (or One-vs-All): \n",
    "            score_value = roc_auc_score(y_true, y_pred, multi_class='ovr', average='macro')\n",
    "            \n",
    "            # Only support average param of: \n",
    "                # 'macro' (default): Calculate metrics for each label, and find their unweighted mean. This does not take label imbalance into account.)\n",
    "                # 'weighted': Calculate metrics for each label, and find their average, weighted by support (the number of true instances for each label).\n",
    "                \n",
    "            # Use average='macro' so model takes extreme imbalance of minority classes into consideration within score.\n",
    "        else:\n",
    "            # AUC uses probabilities. https://scikit-learn.org/stable/modules/generated/sklearn.metrics.roc_auc_score.html\n",
    "            score_value = roc_auc_score(y_true, y_pred)\n",
    "    elif score_metric == 'acc':\n",
    "        # Accuracy score uses threshold value due to ==. https://scikit-learn.org/stable/modules/generated/sklearn.metrics.accuracy_score.html\n",
    "        score_value = accuracy_score(y_true.round(), y_pred.round()) # y_pred needs to be converted explicitly. Dask doesn't know shape of y, y_pred if dask.dataframe.\n",
    "    return(score_value)\n",
    "\n",
    "def dask_xgb_infer(client, run_type, model_file, data_path, label_col, score_metric='auc'):\n",
    "    \"\"\"\n",
    "    Use dask + xgboost for inferencing against already encoded data.\n",
    "    \"\"\"    \n",
    "    if run_type == 'gpu':\n",
    "        import dask_cudf as hw # Needed for xgb.dask.inplace_predict(). \n",
    "#         import dask.dataframe as hw # Has internal switching for compatibility. Takes a long time to perform AUC score, but fast at prediction.\n",
    "    else:\n",
    "        import dask.dataframe as hw\n",
    "    \n",
    "    model = xgb.Booster(model_file=model_file+'.model')\n",
    "\n",
    "    tic = time()\n",
    "    ddf_enc = hw.read_parquet(data_path).persist()\n",
    "\n",
    "    # Format data for xgboost:\n",
    "    feature_cols = [cc for cc in ddf_enc.columns if cc not in excluded_features]\n",
    "#     print(feature_cols)\n",
    "    X = ddf_enc[feature_cols].astype('float32')\n",
    "    y = ddf_enc[label_col].astype('float32')\n",
    "\n",
    "#     # Persist data:\n",
    "#     X = client.persist(X)\n",
    "#     y = client.persist(y)\n",
    "#     wait([X, y])\n",
    "    data_load_time = np.round(time() - tic, 2)\n",
    "\n",
    "    # Run predictions:\n",
    "    tic = time()\n",
    "    model.set_param('predictor', run_type + '_predictor')\n",
    "    model = client.scatter(model, broadcast=True)\n",
    "\n",
    "    # See https://xgboost.readthedocs.io/en/latest/tutorials/dask.html#running-prediction:\n",
    "#     y_pred = xgb.dask.predict(client, model, xgb.dask.DaskDMatrix(client, X, y)) # More efficient to avoid xgb.dask.DaskDMatrix()\n",
    "#     y_pred = xgb.dask.predict(client, model, X) # Use inplace_predict() can sometimes be faster.\n",
    "\n",
    "    # IMPORTANT: X.values required to get consistent CPU scoring. Issue with column ordering in dask.dataframe?\n",
    "    y_pred = xgb.dask.inplace_predict(client, model, X.values) # Use inplace_predict() can sometimes be faster.\n",
    "\n",
    "    # Force computation of y's required for libraries without dask.dataframe support:\n",
    "    y_pred = y_pred.compute()\n",
    "    y = y.compute()\n",
    "    wait([y_pred, y])\n",
    "\n",
    "    predict_time = np.round(time() - tic, 2)\n",
    "\n",
    "    # Score predictions:\n",
    "    print('Scoring predictions in '+ data_path)\n",
    "    tic = time()\n",
    "    score_value = score_pred(run_type, y, y_pred, score_metric)\n",
    "    score_time = np.round(time() - tic, 2)\n",
    "\n",
    "    # TODO: have inference run distributively. Need roc_auc_score equivalent for dask-gpu. Maybe dask_cudf has it?\n",
    "    metrics = {\n",
    "        'data_load_time': data_load_time,\n",
    "        'predict_time': predict_time,\n",
    "        'score':\n",
    "            {score_metric: score_value,\n",
    "            score_metric+'_time': score_time\n",
    "                }\n",
    "        }\n",
    "\n",
    "    del ddf_enc, X, y, y_pred, model\n",
    "    gc.collect()\n",
    "    return(metrics)\n",
    "\n",
    "\n",
    "# Bypass original xgb_model_name to run inference of same model along CPU/GPU code path:\n",
    "# run_type = 'cpu'\n",
    "# xgb_model_name = 'xgb_'+run_type+'_airline_delay_'+study_arpt\n",
    "# xgb_model_name = 'xgb_cpu_airline_delay_'+study_arpt\n",
    "\n",
    "\n",
    "# TODO: Update auc scoring for multi-class problem. cuML roc_auc_score doesn't handle multi-class currently.\n",
    "# https://github.com/rapidsai/cuml/blob/46174b7/python/cuml/metrics/_ranking.py#L119. Should use sklearn directly. Or switch over to xgboost scoring?\n",
    "\n",
    "\n",
    "print('Starting '+run_type+' '+pred_model+' inference rounds....')\n",
    "tic = time()\n",
    "if test_size == 0:\n",
    "    inference_metrics_train = dask_xgb_infer(client, run_type, xgb_model_name, enc_output_dir+'/train', label_col, score_metric=score_metric)\n",
    "    inference_metrics_test = []\n",
    "else:\n",
    "    # Inference against entire data set in two steps:\n",
    "    inference_metrics_train = dask_xgb_infer(client, run_type, xgb_model_name, enc_output_dir+'/train', label_col, score_metric=score_metric)\n",
    "    inference_metrics_test = dask_xgb_infer(client, run_type, xgb_model_name, enc_output_dir+'/test', label_col, score_metric=score_metric)\n",
    "t_dask_infer = np.round(time() - tic, 2)\n",
    "\n",
    "end_time_str = datetime.now().strftime(\"%m/%d/%Y %H:%M:%S\")\n",
    "\n",
    "print('Inference against training data: ', inference_metrics_train)\n",
    "print('Inference against (hold-out) test data: ', inference_metrics_test)\n",
    "print('Total prediction time:', '{:0.2f}'.format(t_dask_infer) + 's')"
   ]
  },
  {
   "cell_type": "raw",
   "id": "4d989437",
   "metadata": {},
   "source": [
    "Full NAS data with 140 attr (includes weather and schedule) 200306-201903 train, 201904-202003 test. Inference requires >48GB, which exceeds single RTX8000.\n",
    "TODO: predict in chunks to reduce mem usage. Then collect predictions and true values for scoring.\n",
    "TODO: find/implement distributed scoring of AUC. Or select a different metric. sklearn auc score single threaded an takes a long time to compute, esp. for multi-class problems."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}