bigquery-samples/dask-bigquery-connector/bigquery_dataproc_dask_xgboost.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "cultural-conversation",
   "metadata": {},
   "source": [
    "# Using Dask and Xgboost with Google Cloud Dataproc\n",
    "\n",
    "In this notebook, you will learn the use of Dask to process dataset from big query leveraging [Dask-BigQuery connector](https://github.com/coiled/dask-bigquery) on Dataptoc.\n",
    "\n",
    "__Dask__ is an open source library for parallel computing written in Python. Dask framework enables us to have a scheduler and a bunch of workers. You submit tasks to the scheduler and it automatically distributes the work among the workers. It works exceptionally well on a single machine, and can scale out to large clusters when needed. \n",
    "\n",
    "__RAPIDS XGBoost__ is a suite of fast, GPU-accelerated Gradient Boost Tree algorithms. \n",
    "\n",
    "__Dataproc__ is a fully managed and highly scalable service for running Apache Spark, Apache Flink, Presto, and 30+ open source tools and frameworks. It is a managed Spark and Hadoop service that lets you take advantage of open source data tools for batch processing, querying, streaming, and machine learning. Dataproc automation helps you create clusters quickly, manage them easily, and save money by turning clusters off when you don't need them. With less time and money spent on administration, you can focus on your jobs and your data. \n",
    "\n",
    "We can create GPU dataproc cluster with following commands:\n",
    "```\n",
    "export REGION=europe-west4\n",
    "export GCS_BUCKET=dongm-tlt\n",
    "export CLUSTER_NAME=dongm-bq\n",
    "export NUM_GPUS=2\n",
    "export NUM_WORKERS=2\n",
    "\n",
    "gcloud dataproc clusters create $CLUSTER_NAME  \\\n",
    "    --region $REGION \\\n",
    "    --zone europe-west4-c \\\n",
    "    --image-version=2.0-ubuntu18 \\\n",
    "    --master-machine-type n1-standard-8 \\\n",
    "    --master-accelerator type=nvidia-tesla-t4,count=1 \\\n",
    "    --num-workers $NUM_WORKERS \\\n",
    "    --worker-accelerator type=nvidia-tesla-t4,count=$NUM_GPUS \\\n",
    "    --worker-machine-type n1-highmem-8 \\\n",
    "    --initialization-actions gs://goog-dataproc-initialization-actions-${REGION}/gpu/install_gpu_driver.sh,gs://goog-dataproc-initialization-actions-${REGION}/dask/dask.sh,gs://goog-dataproc-initialization-actions-${REGION}/rapids/rapids.sh \\\n",
    "    --optional-components=JUPYTER,ZEPPELIN \\\n",
    "    --metadata rapids-runtime=DASK \\\n",
    "    --initialization-action-timeout=60m \\\n",
    "    --bucket $GCS_BUCKET \\\n",
    "    --enable-component-gateway \\\n",
    "    --subnet default\n",
    "```\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "recovered-extension",
   "metadata": {},
   "source": [
    "## Introduction\n",
    "\n",
    "GPUs can greatly accelerate all stages of an ML pipeline: pre-processing, training, and inference. In this workshop, we will be focusing on the pre-processing and training stages, using Python in a Jupyter Notebook environment. First, we will use Dask/RAPIDS to read a dataset into NVIDIA GPU memory and execute some basic functions. Then, we’ll use Dask to scale beyond our GPU memory capacity.\n",
    "\n",
    "This notebook has following sections:\n",
    "\n",
    "* Cluster set up with Dataproc\n",
    "* Introduction to Dask\n",
    "* Data loading from Big Query\n",
    "* ETL with Dataframe operations\n",
    "* Machine Learning with XGBoost\n",
    "\n",
    "Before we begin the notebook, we need to preload the data into a big query table, we can follow Big Query UI to upload NYC Taxi data with wild card: `gs://anaconda-public-data/nyc-taxi/nyc.parquet/part*0.parquet`, this will read 16 parquet files to a big query table, total ~ 1.6GB parquet data."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aggregate-promotion",
   "metadata": {},
   "source": [
    "## Introduction to Dask\n",
    "\n",
    "Dask is the most commonly used parallelism framework within the PyData and SciPy communities. Dask is designed to scale from parallelizing workloads on the CPUs in your laptop to thousands of nodes in a cloud cluster. In conjunction with the open-source RAPIDS framework developed by NVIDIA, you can utilize the parallel processing power of both CPUs and NVIDIA GPUs. \n",
    "\n",
    "In Dask programming, we create computational graphs that define code we **would like** to execute, and then, give these computational graphs to a Dask scheduler which evaluates them lazily, and efficiently, in parallel.\n",
    "\n",
    "In addition to using multiple CPU cores or threads to execute computational graphs in parallel, Dask schedulers can also be configured to execute computational graphs on multiple CPUs, or, as we will do in this workshop, multiple GPUs. As a result, Dask programming facilitates operating on datasets that are larger than the memory of a single compute resource."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "hollow-platform",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "import cupy as cp\n",
    "import numpy as np\n",
    "import os\n",
    "import time\n",
    "\n",
    "# Dask imports\n",
    "import dask\n",
    "import dask_cudf\n",
    "import dask_ml\n",
    "import dask_bigquery\n",
    "\n",
    "from dask.distributed import WorkerPlugin, wait, progress\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "affected-latin",
   "metadata": {},
   "source": [
    "## Nvidia GPUs\n",
    "Let us start by checking hardware available, we can run following command on worker node.\n",
    "```\n",
    "watch -n 0.5 nvidia-smi\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "shaped-moderator",
   "metadata": {},
   "source": [
    "### Starting a `YarnCluster`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "competent-manner",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "21/09/24 21:27:14 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "21/09/24 21:27:15 WARN shortcircuit.DomainSocketFactory: The short-circuit local reads feature cannot be used because libhadoop cannot be loaded.\n",
      "21/09/24 21:27:15 INFO client.RMProxy: Connecting to ResourceManager at cluster-c6ce-telco-dask2-m/10.128.0.8:8032\n",
      "21/09/24 21:27:15 INFO client.AHSProxy: Connecting to Application History server at cluster-c6ce-telco-dask2-m/10.128.0.8:10200\n",
      "21/09/24 21:27:16 INFO skein.Driver: Driver started, listening on 42335\n",
      "21/09/24 21:27:16 INFO conf.Configuration: found resource resource-types.xml at file:/etc/hadoop/conf.empty/resource-types.xml\n",
      "21/09/24 21:27:16 INFO resource.ResourceUtils: Adding resource type - name = yarn.io/gpu, units = , type = COUNTABLE\n",
      "21/09/24 21:27:16 INFO skein.Driver: Uploading application resources to hdfs://cluster-c6ce-telco-dask2-m/user/root/.skein/application_1632504781433_0009\n",
      "21/09/24 21:27:16 INFO skein.Driver: Submitting application...\n",
      "21/09/24 21:27:17 INFO impl.YarnClientImpl: Submitted application application_1632504781433_0009\n"
     ]
    }
   ],
   "source": [
    "from dask.distributed import Client\n",
    "from dask_yarn import YarnCluster\n",
    "\n",
    "cluster = YarnCluster(worker_class=\"dask_cuda.CUDAWorker\", \n",
    "    worker_gpus=1, worker_vcores=4, worker_memory='24GB', \n",
    "    worker_env={\"CONDA_PREFIX\":\"/opt/conda/default/\"})\n",
    "cluster.scale(4)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bulgarian-sunglasses",
   "metadata": {},
   "source": [
    "## Instantiating a Client Connection\n",
    "The `dask.distributed` library gives us distributed functionality, including the ability to connect to the CUDA Cluster we just created. The `progress` import will give us a handy progress bar we can utilize below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "south-yukon",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table style=\"border: 2px solid white;\">\n",
       "<tr>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Client</h3>\n",
       "<ul style=\"text-align: left; list-style: none; margin: 0; padding: 0;\">\n",
       "  <li><b>Scheduler: </b>tcp://10.128.0.9:39345</li>\n",
       "  <li><b>Dashboard: </b><a href='http://10.128.0.9:35653/status' target='_blank'>http://10.128.0.9:35653/status</a></li>\n",
       "</ul>\n",
       "</td>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Cluster</h3>\n",
       "<ul style=\"text-align: left; list-style:none; margin: 0; padding: 0;\">\n",
       "  <li><b>Workers: </b>2</li>\n",
       "  <li><b>Cores: </b>8</li>\n",
       "  <li><b>Memory: </b>52.15 GiB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: 'tcp://10.128.0.9:39345' processes=2 threads=8, memory=52.15 GiB>"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "swiss-whole",
   "metadata": {},
   "source": [
    "In a distrubuted cluster environment, you can choose number of workers and cores based on the cluster resources. Dask ships with a very helpful dashboard that in our case runs on port `8787`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "appreciated-dubai",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "GPU Cluster status:  running\n"
     ]
    }
   ],
   "source": [
    "print('GPU Cluster status: ', client.status)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "legitimate-switch",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of GPU workers:  2\n"
     ]
    }
   ],
   "source": [
    "# Query the client for all connected workers\n",
    "# Number of workers in out cluster\n",
    "workers = client.has_what().keys()\n",
    "n_workers = len(workers)\n",
    "print('Number of GPU workers: ', n_workers)\n",
    "\n",
    "ddf = dask_bigquery.read_gbq(\n",
    "    project_id=\"k80-exploration\",\n",
    "    dataset_id=\"spark_rapids\",\n",
    "    table_id=\"nyc_taxi_0\",\n",
    ")\n",
    "\n",
    "ddf.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "endless-gauge",
   "metadata": {},
   "source": [
    "### Dataset\n",
    "\n",
    "We are using [NYC Taxi Trip Duration Dataset from Kaggle](https://www.kaggle.com/c/nyc-taxi-trip-duration).\n",
    "\n",
    "### Data fields\n",
    "\n",
    "| Colonne            | Description                                                                                                                                                                                                           |\n",
    "|:-------------------|:----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|\n",
    "| __id__                 | a unique identifier for each trip                                                                                                                                                                                     |\n",
    "| __vendor_id__         | a code indicating the provider associated with the trip record                                                                                                                                                        |\n",
    "| __pickup_datetime__    | date and time when the meter was engaged                                                                                                                                                                              |\n",
    "| __dropoff_datetime__   | date and time when the meter was disengaged                                                                                                                                                                           |\n",
    "| __passenger_count__    | the number of passengers in the vehicle (driver entered value)                                                                                                                                                        |\n",
    "| __pickup_longitude__   | the longitude where the meter was engaged                                                                                                                                                                             |\n",
    "| __pickup_latitude__    | the latitude where the meter was engaged                                                                                                                                                                              |\n",
    "| __dropoff_longitude__  | the longitude where the meter was disengaged                                                                                                                                                                          |\n",
    "| __dropoff_latitude__   | the latitude where the meter was disengaged                                                                                                                                                                           |\n",
    "| __store_and_fwd_flag__ | This flag indicates whether the trip record was held in vehicle memory before sending to the vendor because the vehicle did not have a connection to the server (Y=store and forward; N=not a store and forward trip) |\n",
    "| __trip_duration__      | duration of the trip in seconds                                                                                                                                                                                       |"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "presidential-drawing",
   "metadata": {},
   "source": [
    "## ETL Exploration"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "secret-clear",
   "metadata": {},
   "source": [
    "### Taxi Data Configuration (Medium)\n",
    "We can use the parquet data from the anaconda public repo here. Which will illustrate how much faster it is to read parquet, and gives us around 150 million rows of data to work with."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "quiet-postcard",
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean(df_part, remap, must_haves):\n",
    "    \"\"\"\n",
    "    This function performs the various clean up tasks for the data\n",
    "    and returns the cleaned dataframe.\n",
    "    \"\"\"\n",
    "    tmp = {col:col.strip().lower() for col in list(df_part.columns)}\n",
    "    df_part = df_part.rename(columns=tmp)\n",
    "    \n",
    "    # rename using the supplied mapping\n",
    "    df_part = df_part.rename(columns=remap)\n",
    "    \n",
    "    # iterate through columns in this df partition\n",
    "    for col in df_part.columns:\n",
    "        # drop anything not in our expected list\n",
    "        if col not in must_haves:\n",
    "            df_part = df_part.drop(col, axis=1)\n",
    "            continue\n",
    "\n",
    "        # fixes datetime error found by Ty Mckercher and fixed by Paul Mahler\n",
    "        if df_part[col].dtype == 'object' and col in ['pickup_datetime', 'dropoff_datetime']:\n",
    "            df_part[col] = df_part[col].astype('datetime64[ms]')\n",
    "            continue\n",
    "\n",
    "        # if column was read as a string, recast as float\n",
    "        if df_part[col].dtype == 'object':\n",
    "            df_part[col] = df_part[col].astype('float32')\n",
    "        else:\n",
    "            # downcast from 64bit to 32bit types\n",
    "            # Tesla T4 are faster on 32bit ops\n",
    "            if 'int' in str(df_part[col].dtype):\n",
    "                df_part[col] = df_part[col].astype('int32')\n",
    "            if 'float' in str(df_part[col].dtype):\n",
    "                df_part[col] = df_part[col].astype('float32')\n",
    "            df_part[col] = df_part[col].fillna(-1)\n",
    "            \n",
    "    return df_part\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "becoming-pathology",
   "metadata": {},
   "outputs": [],
   "source": [
    "# list of column names that need to be re-mapped\n",
    "remap = {}\n",
    "remap['tpep_pickup_datetime'] = 'pickup_datetime'\n",
    "remap['tpep_dropoff_datetime'] = 'dropoff_datetime'\n",
    "remap['ratecodeid'] = 'rate_code'\n",
    "\n",
    "#create a list of columns & dtypes the df must have\n",
    "must_haves = {\n",
    " 'pickup_datetime': 'datetime64[ms]',\n",
    " 'dropoff_datetime': 'datetime64[ms]',\n",
    " 'passenger_count': 'int32',\n",
    " 'trip_distance': 'float32',\n",
    " 'pickup_longitude': 'float32',\n",
    " 'pickup_latitude': 'float32',\n",
    " 'rate_code': 'int32',\n",
    " 'dropoff_longitude': 'float32',\n",
    " 'dropoff_latitude': 'float32',\n",
    " 'fare_amount': 'float32'\n",
    "}\n",
    "\n",
    "# apply a list of filter conditions to throw out records with missing or outlier values\n",
    "query_frags = [\n",
    "    'fare_amount > 0 and fare_amount < 500',\n",
    "    'passenger_count > 0 and passenger_count < 6',\n",
    "    'pickup_longitude > -75 and pickup_longitude < -73',\n",
    "    'dropoff_longitude > -75 and dropoff_longitude < -73',\n",
    "    'pickup_latitude > 40 and pickup_latitude < 42',\n",
    "    'dropoff_latitude > 40 and dropoff_latitude < 42'\n",
    "]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "realistic-calculation",
   "metadata": {},
   "outputs": [],
   "source": [
    "# This is path to our dataset and it is a parquet file. Can we load a CSV file too? Oh yes! \n",
    "# taxi_parquet_path = \"gs://anaconda-public-data/nyc-taxi/nyc.parquet\"\n",
    "\n",
    "# We are considering a limited columns for our ML pipeline.\n",
    "response_id = 'fare_amount'\n",
    "fields = ['passenger_count', 'trip_distance', 'pickup_longitude', 'pickup_latitude', 'rate_code',\n",
    "             'dropoff_longitude', 'dropoff_latitude', 'fare_amount']\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "outdoor-andorra",
   "metadata": {},
   "source": [
    "### Load Dataset\n",
    "By default, dask uses default block size of 64 MB each. So, `size of dataset / 64` will be number of tasks spread across partitions. If you desire to create a DataFrame with a specific block size, and use `npartitions` argument like below. We used `npartitions=len(workers)` to split data equally into all workers."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "pending-harvey",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.18 s, sys: 677 ms, total: 1.86 s\n",
      "Wall time: 2.13 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "taxi_df = dask_cudf.from_dask_dataframe(ddf)\n",
    "taxi_df = taxi_df.repartition(npartitions=len(workers))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "molecular-honolulu",
   "metadata": {},
   "outputs": [],
   "source": [
    "taxi_df = clean(taxi_df, remap, must_haves)\n",
    "taxi_df = taxi_df.query(' and '.join(query_frags))\n",
    "taxi_df = taxi_df[fields]\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "valid-sacrifice",
   "metadata": {},
   "source": [
    "__Woah!__ I thought we will see data. What do you think has happened here?\n",
    "\n",
    "Because Dask is lazy, the computation has not yet occurred. We can see that there are 160 tasks in the task graph. We can force computation by using `persist`. By forcing execution, the result is now explicitly in memory and our task graph only contains one task per partition (the baseline)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "wrong-harmony",
   "metadata": {},
   "outputs": [],
   "source": [
    "response_dtype=np.float32\n",
    "\n",
    "with dask.annotate(workers=set(workers)):\n",
    "        taxi_df = client.persist(collections=taxi_df)\n",
    "    \n",
    "wait(taxi_df)\n",
    "\n",
    "X = taxi_df[taxi_df.columns.difference([response_id])].astype(np.float32)\n",
    "y = taxi_df[response_id].astype(response_dtype)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "developmental-council",
   "metadata": {},
   "source": [
    "####  Split data to training and test set\n",
    "We will use the `train_test_split()` method from `Dask_ML`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "tender-traveler",
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_ml.model_selection import train_test_split\n",
    "\n",
    "X_train, X_test, y_train, y_test = train_test_split(X, y, \\\n",
    "                                                    train_size=0.80, \\\n",
    "                                                    random_state=42, \\\n",
    "                                                    shuffle=True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "reported-rescue",
   "metadata": {},
   "outputs": [],
   "source": [
    "X_train = X.persist()\n",
    "X_test = X.persist()\n",
    "y_train = y.persist()\n",
    "y_test = y.persist()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "humanitarian-translation",
   "metadata": {},
   "source": [
    "### XGBoost\n",
    "\n",
    "XGBoost falls under the category of Boosting techniques in Ensemble Learning. The algorithm was developed to efficiently reduce computing time and allocate an optimal usage of memory resources. Important features of implementation include handling of missing values (Sparse Aware), Block Structure to support parallelization in tree construction and the ability to fit and boost on new data added to a trained model. ([reference](https://www.kdnuggets.com/2017/10/xgboost-top-machine-learning-method-kaggle-explained.html)) \n",
    "\n",
    "Here is the original paper,\n",
    "[XGBoost: A Scalable Tree Boosting System](https://arxiv.org/abs/1603.02754)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "pleased-greensboro",
   "metadata": {},
   "outputs": [],
   "source": [
    "import xgboost as xgb\n",
    "\n",
    "dmatrix_train = xgb.dask.DaskDMatrix(client, X_train, y_train)\n",
    "dmatrix_test = xgb.dask.DaskDMatrix(client, X_test, y_test)\n",
    "\n",
    "params = {\"max_depth\": 8,\n",
    "        \"max_leaves\": 2 ** 8,\n",
    "        \"alpha\": 0.9,\n",
    "        \"eta\": 0.1,\n",
    "        \"gamma\": 0.1,\n",
    "        \"learning_rate\": 0.1,\n",
    "        \"subsample\": 1,\n",
    "        \"reg_lambda\": 1,\n",
    "        \"scale_pos_weight\": 2,\n",
    "        \"min_child_weight\": 30,\n",
    "        \"tree_method\": \"gpu_hist\", #use GPU for tree building\n",
    "        \"objective\": \"reg:squaredlogerror\",\n",
    "        \"grow_policy\": \"lossguide\",\n",
    "    }\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "gorgeous-nightlife",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 107 ms, sys: 7.04 ms, total: 114 ms\n",
      "Wall time: 13.2 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Use xgboost.dask to distribute across workers-- returns dict \n",
    "xgb_clasf = xgb.dask.train(client, \n",
    "                           params,\n",
    "                           dmatrix_train, \n",
    "                           num_boost_round=2000,\n",
    "                           evals=[(dmatrix_train, 'train'), (dmatrix_test,'test')]\n",
    "                          ) \n",
    "\n",
    "# booster is the trained model \n",
    "model_xgb = xgb_clasf['booster'] \n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "listed-original",
   "metadata": {},
   "source": [
    "### Inferencing\n",
    "#### Measuring model performance\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "chinese-forward",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr>\n",
       "<td>\n",
       "<table>\n",
       "  <thead>\n",
       "    <tr><td> </td><th> Array </th><th> Chunk </th></tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr><th> Bytes </th><td> 6.50 MiB </td> <td> 770.58 kiB </td></tr>\n",
       "    <tr><th> Shape </th><td> (1703313,) </td> <td> (197268,) </td></tr>\n",
       "    <tr><th> Count </th><td> 30 Tasks </td><td> 10 Chunks </td></tr>\n",
       "    <tr><th> Type </th><td> float32 </td><td> numpy.ndarray </td></tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</td>\n",
       "<td>\n",
       "<svg width=\"170\" height=\"75\" style=\"stroke:rgb(0,0,0);stroke-width:1\" >\n",
       "\n",
       "  <!-- Horizontal lines -->\n",
       "  <line x1=\"0\" y1=\"0\" x2=\"120\" y2=\"0\" style=\"stroke-width:2\" />\n",
       "  <line x1=\"0\" y1=\"25\" x2=\"120\" y2=\"25\" style=\"stroke-width:2\" />\n",
       "\n",
       "  <!-- Vertical lines -->\n",
       "  <line x1=\"0\" y1=\"0\" x2=\"0\" y2=\"25\" style=\"stroke-width:2\" />\n",
       "  <line x1=\"11\" y1=\"0\" x2=\"11\" y2=\"25\" />\n",
       "  <line x1=\"22\" y1=\"0\" x2=\"22\" y2=\"25\" />\n",
       "  <line x1=\"35\" y1=\"0\" x2=\"35\" y2=\"25\" />\n",
       "  <line x1=\"49\" y1=\"0\" x2=\"49\" y2=\"25\" />\n",
       "  <line x1=\"60\" y1=\"0\" x2=\"60\" y2=\"25\" />\n",
       "  <line x1=\"74\" y1=\"0\" x2=\"74\" y2=\"25\" />\n",
       "  <line x1=\"84\" y1=\"0\" x2=\"84\" y2=\"25\" />\n",
       "  <line x1=\"93\" y1=\"0\" x2=\"93\" y2=\"25\" />\n",
       "  <line x1=\"106\" y1=\"0\" x2=\"106\" y2=\"25\" />\n",
       "  <line x1=\"120\" y1=\"0\" x2=\"120\" y2=\"25\" style=\"stroke-width:2\" />\n",
       "\n",
       "  <!-- Colored Rectangle -->\n",
       "  <polygon points=\"0.0,0.0 120.0,0.0 120.0,25.412616514582485 0.0,25.412616514582485\" style=\"fill:#ECB172A0;stroke-width:0\"/>\n",
       "\n",
       "  <!-- Text -->\n",
       "  <text x=\"60.000000\" y=\"45.412617\" font-size=\"1.0rem\" font-weight=\"100\" text-anchor=\"middle\" >1703313</text>\n",
       "  <text x=\"140.000000\" y=\"12.706308\" font-size=\"1.0rem\" font-weight=\"100\" text-anchor=\"middle\" transform=\"rotate(0,140.000000,12.706308)\">1</text>\n",
       "</svg>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "dask.array<concatenate, shape=(1703313,), dtype=float32, chunksize=(197268,), chunktype=numpy.ndarray>"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Set to use GPU for inference.\n",
    "model_xgb.set_param({'predictor': 'gpu_predictor'})\n",
    "# dtrain is the DaskDMatrix defined above.\n",
    "y_predict = xgb.dask.predict(client, xgb_clasf['booster'], dmatrix_test)\n",
    "y_predict\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "7fb50645-b33a-4805-88d8-1ff73d7f98c7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Preditction Labels are of type : <class 'dask.array.core.Array'>\n",
      "Test Labels are of type : <class 'dask_cudf.core.DataFrame'>\n"
     ]
    }
   ],
   "source": [
    "print(\"Preditction Labels are of type : \" + str(type(y_predict)))\n",
    "print(\"Test Labels are of type : \" + str(type(y_test)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "08389fd3-0700-478b-b2b3-9a3b294d324a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([58.520275 ,  8.61709  ,  6.8902235, ..., 15.131949 , 21.359398 ,\n",
       "       20.86175  ], dtype=float32)"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "y = cp.asarray(y_predict)\n",
    "y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "d24f1149-6776-4641-b2ad-474f4d78296b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[69.5],\n",
       "       [ 8. ],\n",
       "       [ 5.7],\n",
       "       ...,\n",
       "       [14.5],\n",
       "       [24. ],\n",
       "       [20.5]], dtype=float32)"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "t = y_test.to_dask_array()\n",
    "t = t.compute()\n",
    "t"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "elementary-vintage",
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask\n",
    "\n",
    "y_t, y_p = dask.compute(t, y)\n",
    "#y_test_c, y_predict = dask.compute(y_test, y_predict)\n",
    "print(\"XGB model AUC: {:.2%}\".format(cuml.metrics.regression.mean_absolute_error(y_t, y_p, multioutput='uniform_average')))\n",
    "\n",
    "#print(\"XGB model AUC: {:.2%}\".format(cuml.metrics.roc_auc_score(y_t, y_p)))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "sized-outreach",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "XGBoost model accuracy:   36.46%\n"
     ]
    }
   ],
   "source": [
    "print(\"XGBoost model accuracy:   {:.2%}\".format(cuml.metrics.accuracy_score(y_t, y_p)))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "accessible-physics",
   "metadata": {},
   "source": [
    "#### Resources\n",
    "\n",
    "##### Dask\n",
    "\n",
    "##### XGBoost\n",
    "* [Ensemble Learning to Improve Machine Learning Results](https://blog.statsbot.co/ensemble-learning-d1dcd548e936)\n",
    "* [Complete Guide to Parameter Tuning in XGBoost with codes in Python](https://www.analyticsvidhya.com/blog/2016/03/complete-guide-parameter-tuning-xgboost-with-codes-python/)\n",
    "* [Understanding XGBoost Algorithm In Detail](https://analyticsindiamag.com/xgboost-internal-working-to-make-decision-trees-and-deduce-predictions/)\n",
    "\n",
    "##### Random Forest Regressor\n",
    "* [Random Forest](https://williamkoehrsen.medium.com/random-forest-simple-explanation-377895a60d2d) \\\n",
    "* [Random Forest Regression](https://towardsdatascience.com/machine-learning-basics-random-forest-regression-be3e1e3bb91a)\n",
    "* [Classification and Regression by randomForest](https://www.researchgate.net/profile/Andy-Liaw/publication/228451484_Classification_and_Regression_by_RandomForest/links/53fb24cc0cf20a45497047ab/Classification-and-Regression-by-RandomForest.pdf)\n"
   ]
  }
 ],
 "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.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}