03_Dask_ML-LargeDS.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 1.  Pre-processing in Pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Note that this code must not run directly the Workbench in the graphical console. \n",
    "# Use python3, ipython3 or Jupyter notebook in a session's terminal, \n",
    "# See https://github.com/dask/dask/issues/4612\n",
    "\n",
    "import cdsw_dask_utils\n",
    "import cdsw\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Month</th>\n",
       "      <th>DayofMonth</th>\n",
       "      <th>DayOfWeek</th>\n",
       "      <th>CRSDepTime</th>\n",
       "      <th>CRSArrTime</th>\n",
       "      <th>UniqueCarrier</th>\n",
       "      <th>FlightNum</th>\n",
       "      <th>Origin</th>\n",
       "      <th>Dest</th>\n",
       "      <th>Cancelled</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>9</td>\n",
       "      <td>6</td>\n",
       "      <td>1331</td>\n",
       "      <td>1435</td>\n",
       "      <td>PI</td>\n",
       "      <td>942</td>\n",
       "      <td>SYR</td>\n",
       "      <td>BWI</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>10</td>\n",
       "      <td>7</td>\n",
       "      <td>1331</td>\n",
       "      <td>1435</td>\n",
       "      <td>PI</td>\n",
       "      <td>942</td>\n",
       "      <td>SYR</td>\n",
       "      <td>BWI</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1</td>\n",
       "      <td>11</td>\n",
       "      <td>1</td>\n",
       "      <td>1331</td>\n",
       "      <td>1435</td>\n",
       "      <td>PI</td>\n",
       "      <td>942</td>\n",
       "      <td>SYR</td>\n",
       "      <td>BWI</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1</td>\n",
       "      <td>12</td>\n",
       "      <td>2</td>\n",
       "      <td>1331</td>\n",
       "      <td>1435</td>\n",
       "      <td>PI</td>\n",
       "      <td>942</td>\n",
       "      <td>SYR</td>\n",
       "      <td>BWI</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1</td>\n",
       "      <td>13</td>\n",
       "      <td>3</td>\n",
       "      <td>1331</td>\n",
       "      <td>1435</td>\n",
       "      <td>PI</td>\n",
       "      <td>942</td>\n",
       "      <td>SYR</td>\n",
       "      <td>BWI</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   Month  DayofMonth  DayOfWeek  CRSDepTime  CRSArrTime UniqueCarrier  \\\n",
       "0      1           9          6        1331        1435            PI   \n",
       "1      1          10          7        1331        1435            PI   \n",
       "2      1          11          1        1331        1435            PI   \n",
       "3      1          12          2        1331        1435            PI   \n",
       "4      1          13          3        1331        1435            PI   \n",
       "\n",
       "   FlightNum Origin Dest  Cancelled  \n",
       "0        942    SYR  BWI          0  \n",
       "1        942    SYR  BWI          0  \n",
       "2        942    SYR  BWI          0  \n",
       "3        942    SYR  BWI          0  \n",
       "4        942    SYR  BWI          0  "
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# read flights dataset\n",
    "input_file = \"data/1988.csv.bz2\"\n",
    "columns = ['Month','DayofMonth','DayOfWeek','CRSDepTime','CRSArrTime','UniqueCarrier','FlightNum',\n",
    "           'Origin','Dest','Cancelled']\n",
    "flight_df = pd.read_csv(input_file,sep=',',header=0, na_values='NA', usecols=columns )\n",
    "flight_df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 5202096 entries, 0 to 5202095\n",
      "Data columns (total 10 columns):\n",
      "Month            int64\n",
      "DayofMonth       int64\n",
      "DayOfWeek        int64\n",
      "CRSDepTime       int64\n",
      "CRSArrTime       int64\n",
      "UniqueCarrier    object\n",
      "FlightNum        int64\n",
      "Origin           object\n",
      "Dest             object\n",
      "Cancelled        int64\n",
      "dtypes: int64(7), object(3)\n",
      "memory usage: 396.9+ MB\n"
     ]
    }
   ],
   "source": [
    "flight_df.info()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Feature Engineering\n",
    "#### 1. create routes from origin and destination"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "LAX_SFO    20750\n",
       "SFO_LAX    20658\n",
       "LAX_PHX    13461\n",
       "PHX_LAX    13273\n",
       "LAX_LAS    12175\n",
       "LGA_BOS    12027\n",
       "LAS_LAX    11801\n",
       "SJC_LAX    11535\n",
       "LAX_SJC    11292\n",
       "BOS_LGA    11141\n",
       "Name: route, dtype: int64"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import numpy as np\n",
    "# feature engineering - Create routes\n",
    "flight_df['route'] = flight_df['Origin'] + \"_\" + flight_df['Dest']\n",
    "\n",
    "# show top 20 routes - As defined by nb of flights\n",
    "top_routes = flight_df['route'].value_counts(ascending=False)\n",
    "top_routes.head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>route</th>\n",
       "      <th>count</th>\n",
       "      <th>nb_cancelled</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>1920</th>\n",
       "      <td>LAX_SFO</td>\n",
       "      <td>20750</td>\n",
       "      <td>228</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3375</th>\n",
       "      <td>SFO_LAX</td>\n",
       "      <td>20658</td>\n",
       "      <td>206</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1911</th>\n",
       "      <td>LAX_PHX</td>\n",
       "      <td>13461</td>\n",
       "      <td>78</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2870</th>\n",
       "      <td>PHX_LAX</td>\n",
       "      <td>13273</td>\n",
       "      <td>71</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1894</th>\n",
       "      <td>LAX_LAS</td>\n",
       "      <td>12175</td>\n",
       "      <td>58</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1956</th>\n",
       "      <td>LGA_BOS</td>\n",
       "      <td>12027</td>\n",
       "      <td>287</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1840</th>\n",
       "      <td>LAS_LAX</td>\n",
       "      <td>11801</td>\n",
       "      <td>47</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3432</th>\n",
       "      <td>SJC_LAX</td>\n",
       "      <td>11535</td>\n",
       "      <td>71</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1921</th>\n",
       "      <td>LAX_SJC</td>\n",
       "      <td>11292</td>\n",
       "      <td>71</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>366</th>\n",
       "      <td>BOS_LGA</td>\n",
       "      <td>11141</td>\n",
       "      <td>243</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        route  count  nb_cancelled\n",
       "1920  LAX_SFO  20750           228\n",
       "3375  SFO_LAX  20658           206\n",
       "1911  LAX_PHX  13461            78\n",
       "2870  PHX_LAX  13273            71\n",
       "1894  LAX_LAS  12175            58\n",
       "1956  LGA_BOS  12027           287\n",
       "1840  LAS_LAX  11801            47\n",
       "3432  SJC_LAX  11535            71\n",
       "1921  LAX_SJC  11292            71\n",
       "366   BOS_LGA  11141           243"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#look at their cancelations\n",
    "flight_df[['route', 'Cancelled']].groupby('route')\\\n",
    "        .agg({'route':'size', 'Cancelled':'sum'}) \\\n",
    "        .rename(columns={'route':'count','Cancelled':'nb_cancelled'}) \\\n",
    "        .reset_index()\\\n",
    "        .sort_values(['count'],ascending=False)\\\n",
    "        .head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "Int64Index: 487253 entries, 786 to 5201496\n",
      "Data columns (total 11 columns):\n",
      "Month            487253 non-null int64\n",
      "DayofMonth       487253 non-null int64\n",
      "DayOfWeek        487253 non-null int64\n",
      "CRSDepTime       487253 non-null int64\n",
      "CRSArrTime       487253 non-null int64\n",
      "UniqueCarrier    487253 non-null object\n",
      "FlightNum        487253 non-null int64\n",
      "Origin           487253 non-null object\n",
      "Dest             487253 non-null object\n",
      "Cancelled        487253 non-null int64\n",
      "route            487253 non-null object\n",
      "dtypes: int64(7), object(4)\n",
      "memory usage: 44.6+ MB\n"
     ]
    }
   ],
   "source": [
    "#focus on 50 biggest routes - As defined by nb of flights \n",
    "route_lst=top_routes.head(50)\n",
    "flight_df = flight_df[flight_df['route'].isin(route_lst.index)]\n",
    "flight_df.info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Month            0\n",
       "DayofMonth       0\n",
       "DayOfWeek        0\n",
       "CRSDepTime       0\n",
       "CRSArrTime       0\n",
       "UniqueCarrier    0\n",
       "FlightNum        0\n",
       "Origin           0\n",
       "Dest             0\n",
       "Cancelled        0\n",
       "route            0\n",
       "dtype: int64"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Quick sanity check - count number of null values()\n",
    "flight_df.isnull().sum()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2. Feature and label encoding encoding\n",
    "Label encoding"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# ### encode labels \n",
    "flight_df.Cancelled = pd.Categorical(flight_df.Cancelled)\n",
    "flight_df['Label'] = flight_df.Cancelled.cat.codes\n",
    "flight_df.drop(['Cancelled'], axis=1, inplace=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Feature encoding "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array(['ATL', 'BOS', 'DAL', 'DCA', 'DEN', 'DFW', 'DTW', 'EWR', 'HOU',\n",
       "       'IAH', 'LAS', 'LAX', 'LGA', 'MCO', 'MIA', 'MSP', 'ORD', 'PDX',\n",
       "       'PHX', 'SAN', 'SAT', 'SEA', 'SFO', 'SJC', 'STL'], dtype=object)"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Get airport unique values\n",
    "import numpy as np\n",
    "airport_list = np.unique(np.concatenate((np.sort(flight_df['Origin'].unique()),\n",
    "                                         np.sort(flight_df['Dest'].unique())))\n",
    "                        )\n",
    "airport_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Encode categorical features - Sklearn only supports numerical values\n",
    "from sklearn.preprocessing import LabelEncoder\n",
    "\n",
    "# encode airlines \n",
    "le_carrier = LabelEncoder()\n",
    "flight_df['Carrier_encoded'] = le_carrier.fit_transform(flight_df['UniqueCarrier'])\n",
    "\n",
    "# Encode airports : Using same encoder for both origin and dest ( consistent encoding of airports )\n",
    "le_airport = LabelEncoder()\n",
    "le_airport.fit(airport_list)\n",
    "flight_df['Origin_encoded'] = le_airport.transform(flight_df['Origin'])\n",
    "flight_df['Dest_encoded'] = le_airport.transform(flight_df['Dest'])\n",
    "\n",
    "# Encode routes \n",
    "le_route = LabelEncoder()\n",
    "flight_df['route_encoded'] = le_route.fit_transform(flight_df['route'])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>UniqueCarrier</th>\n",
       "      <th>Carrier_encoded</th>\n",
       "      <th>Origin</th>\n",
       "      <th>Origin_encoded</th>\n",
       "      <th>Dest</th>\n",
       "      <th>Dest_encoded</th>\n",
       "      <th>route</th>\n",
       "      <th>route_encoded</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>1860161</th>\n",
       "      <td>HP</td>\n",
       "      <td>5</td>\n",
       "      <td>PHX</td>\n",
       "      <td>18</td>\n",
       "      <td>LAX</td>\n",
       "      <td>11</td>\n",
       "      <td>PHX_LAX</td>\n",
       "      <td>36</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2265131</th>\n",
       "      <td>WN</td>\n",
       "      <td>13</td>\n",
       "      <td>LAS</td>\n",
       "      <td>10</td>\n",
       "      <td>PHX</td>\n",
       "      <td>18</td>\n",
       "      <td>LAS_PHX</td>\n",
       "      <td>15</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>932249</th>\n",
       "      <td>UA</td>\n",
       "      <td>11</td>\n",
       "      <td>LAX</td>\n",
       "      <td>11</td>\n",
       "      <td>SFO</td>\n",
       "      <td>22</td>\n",
       "      <td>LAX_SFO</td>\n",
       "      <td>20</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3888083</th>\n",
       "      <td>US</td>\n",
       "      <td>12</td>\n",
       "      <td>SFO</td>\n",
       "      <td>22</td>\n",
       "      <td>SEA</td>\n",
       "      <td>21</td>\n",
       "      <td>SFO_SEA</td>\n",
       "      <td>47</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1865490</th>\n",
       "      <td>HP</td>\n",
       "      <td>5</td>\n",
       "      <td>LAX</td>\n",
       "      <td>11</td>\n",
       "      <td>PHX</td>\n",
       "      <td>18</td>\n",
       "      <td>LAX_PHX</td>\n",
       "      <td>17</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2266661</th>\n",
       "      <td>EA</td>\n",
       "      <td>4</td>\n",
       "      <td>BOS</td>\n",
       "      <td>1</td>\n",
       "      <td>DCA</td>\n",
       "      <td>3</td>\n",
       "      <td>BOS_DCA</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3308115</th>\n",
       "      <td>DL</td>\n",
       "      <td>3</td>\n",
       "      <td>PHX</td>\n",
       "      <td>18</td>\n",
       "      <td>LAS</td>\n",
       "      <td>10</td>\n",
       "      <td>PHX_LAS</td>\n",
       "      <td>35</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1641615</th>\n",
       "      <td>AA</td>\n",
       "      <td>0</td>\n",
       "      <td>IAH</td>\n",
       "      <td>9</td>\n",
       "      <td>DFW</td>\n",
       "      <td>5</td>\n",
       "      <td>IAH_DFW</td>\n",
       "      <td>13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4636558</th>\n",
       "      <td>EA</td>\n",
       "      <td>4</td>\n",
       "      <td>BOS</td>\n",
       "      <td>1</td>\n",
       "      <td>LGA</td>\n",
       "      <td>12</td>\n",
       "      <td>BOS_LGA</td>\n",
       "      <td>3</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4735793</th>\n",
       "      <td>CO</td>\n",
       "      <td>2</td>\n",
       "      <td>EWR</td>\n",
       "      <td>7</td>\n",
       "      <td>BOS</td>\n",
       "      <td>1</td>\n",
       "      <td>EWR_BOS</td>\n",
       "      <td>10</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        UniqueCarrier  Carrier_encoded Origin  Origin_encoded Dest  \\\n",
       "1860161            HP                5    PHX              18  LAX   \n",
       "2265131            WN               13    LAS              10  PHX   \n",
       "932249             UA               11    LAX              11  SFO   \n",
       "3888083            US               12    SFO              22  SEA   \n",
       "1865490            HP                5    LAX              11  PHX   \n",
       "2266661            EA                4    BOS               1  DCA   \n",
       "3308115            DL                3    PHX              18  LAS   \n",
       "1641615            AA                0    IAH               9  DFW   \n",
       "4636558            EA                4    BOS               1  LGA   \n",
       "4735793            CO                2    EWR               7  BOS   \n",
       "\n",
       "         Dest_encoded    route  route_encoded  \n",
       "1860161            11  PHX_LAX             36  \n",
       "2265131            18  LAS_PHX             15  \n",
       "932249             22  LAX_SFO             20  \n",
       "3888083            21  SFO_SEA             47  \n",
       "1865490            18  LAX_PHX             17  \n",
       "2266661             3  BOS_DCA              1  \n",
       "3308115            10  PHX_LAS             35  \n",
       "1641615             5  IAH_DFW             13  \n",
       "4636558            12  BOS_LGA              3  \n",
       "4735793             1  EWR_BOS             10  "
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Viewing few rows of make and its encoded columns\n",
    "flight_df[['UniqueCarrier','Carrier_encoded','Origin','Origin_encoded',\n",
    "           'Dest', 'Dest_encoded', 'route', 'route_encoded' ]].sample(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# # 2. Build a classification model using MLLib\n",
    "# ### Split Test/Train\n",
    "\n",
    "from sklearn.model_selection import train_test_split\n",
    "X_train, X_test, y_train, y_test = train_test_split(flight_df.drop(['UniqueCarrier','Origin','Dest','route'],axis=1),\n",
    "                                                    flight_df['Label'], \n",
    "                                                    test_size=0.3, \n",
    "                                                    random_state=100)\n",
    "del flight_df # free up some memory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "Int64Index: 341077 entries, 936357 to 3164854\n",
      "Data columns (total 11 columns):\n",
      "Month              341077 non-null int64\n",
      "DayofMonth         341077 non-null int64\n",
      "DayOfWeek          341077 non-null int64\n",
      "CRSDepTime         341077 non-null int64\n",
      "CRSArrTime         341077 non-null int64\n",
      "FlightNum          341077 non-null int64\n",
      "Label              341077 non-null int8\n",
      "Carrier_encoded    341077 non-null int64\n",
      "Origin_encoded     341077 non-null int64\n",
      "Dest_encoded       341077 non-null int64\n",
      "route_encoded      341077 non-null int64\n",
      "dtypes: int64(10), int8(1)\n",
      "memory usage: 28.9 MB\n"
     ]
    }
   ],
   "source": [
    "X_train.info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "Int64Index: 146176 entries, 1856532 to 3419091\n",
      "Data columns (total 11 columns):\n",
      "Month              146176 non-null int64\n",
      "DayofMonth         146176 non-null int64\n",
      "DayOfWeek          146176 non-null int64\n",
      "CRSDepTime         146176 non-null int64\n",
      "CRSArrTime         146176 non-null int64\n",
      "FlightNum          146176 non-null int64\n",
      "Label              146176 non-null int8\n",
      "Carrier_encoded    146176 non-null int64\n",
      "Origin_encoded     146176 non-null int64\n",
      "Dest_encoded       146176 non-null int64\n",
      "route_encoded      146176 non-null int64\n",
      "dtypes: int64(10), int8(1)\n",
      "memory usage: 12.4 MB\n"
     ]
    }
   ],
   "source": [
    "X_test.info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 2 : Model Training - Using SKLEARN "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.ensemble import RandomForestClassifier\n",
    "from sklearn.model_selection import GridSearchCV\n",
    "\n",
    "# ### parameters for grid search\n",
    "param_numTrees = list(range(2,12,2))\n",
    "print(param_numTrees)\n",
    "param_maxDepth = list(range(2,12,2))\n",
    "print(param_maxDepth)\n",
    "\n",
    "rfc = RandomForestClassifier(random_state=10, n_jobs=-1)\n",
    "\n",
    "GS_params = { \n",
    "    'n_estimators': param_numTrees,\n",
    "    'max_depth' : param_maxDepth\n",
    "}\n",
    "\n",
    "CV_rfc = GridSearchCV(estimator=rfc, \n",
    "                      param_grid=GS_params, \n",
    "                      cv= 3,\n",
    "                      verbose = 1,\n",
    "                      n_jobs=-1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "## Train model\n",
    "CV_rfc.fit(X_train, y_train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'max_depth': 2, 'n_estimators': 2}\n"
     ]
    }
   ],
   "source": [
    "# ### Show Best Parameters \n",
    "print(CV_rfc.best_params_)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Evaluate model performance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.metrics import classification_report\n",
    "# ## fit test dataset \n",
    "y_true, y_pred = y_test, CV_rfc.predict(X_test)\n",
    "# ## Print classification metrics \n",
    "print(classification_report(y_true, y_pred))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 3. Distributed training using DASK Backend"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Waiting for Dask scheduler to become ready...\n",
      "Dask scheduler is ready\n",
      "IDs ['gxyld8kqktr76c0o', 'ltw6rul8aqawl8mj']\n"
     ]
    }
   ],
   "source": [
    "# Run a Dask cluster with three workers and return an object containing\n",
    "# a description of the cluster. \n",
    "# \n",
    "# Using helper library \n",
    "#\n",
    "# Note that the scheduler will run in the current session, and the Dask\n",
    "# dashboard will become available in the nine-dot menu at the upper\n",
    "# right corner of the CDSW app.\n",
    "\n",
    "cluster = cdsw_dask_utils.run_dask_cluster(\n",
    "  n=2, \\\n",
    "  cpu=1, \\\n",
    "  memory=3, \\\n",
    "  nvidia_gpu=0\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<a  target=\"_blank\" rel=\"noopener noreferrer\" href=\"http://read-only-686th6hcdcx8wn73.mlamairesse-webinar-4.vpc.cloudera.com\">http://read-only-686th6hcdcx8wn73.mlamairesse-webinar-4.vpc.cloudera.com</a>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Get the Dask Scheduler UI\n",
    "import os \n",
    "engine_id = os.environ.get('CDSW_ENGINE_ID')\n",
    "cdsw_domain = os.environ.get('CDSW_DOMAIN')\n",
    "\n",
    "from IPython.core.display import HTML\n",
    "HTML('<a  target=\"_blank\" rel=\"noopener noreferrer\" href=\"http://read-only-{}.{}\">http://read-only-{}.{}</a>'\n",
    "     .format(engine_id,cdsw_domain,engine_id,cdsw_domain))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# #### Connect a Dask client to the scheduler address in the cluster\n",
    "\n",
    "from dask.distributed import Client\n",
    "client = Client(cluster[\"scheduler_address\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2, 4, 6, 8, 10]\n",
      "[2, 4, 6, 8, 10]\n"
     ]
    }
   ],
   "source": [
    "from sklearn.ensemble import RandomForestClassifier\n",
    "from sklearn.model_selection import GridSearchCV\n",
    "\n",
    "# ### parameters for grid search\n",
    "param_numTrees = list(range(2,12,2))\n",
    "print(param_numTrees)\n",
    "param_maxDepth = list(range(2,12,2))\n",
    "print(param_maxDepth)\n",
    "\n",
    "rfc = RandomForestClassifier(random_state=10, n_jobs=-1)\n",
    "\n",
    "GS_params = { \n",
    "    'n_estimators': param_numTrees,\n",
    "    'max_depth' : param_maxDepth\n",
    "}\n",
    "\n",
    "CV_rfc = GridSearchCV(estimator=rfc, \n",
    "                      param_grid=GS_params, \n",
    "                      cv= 3,\n",
    "                      verbose = 1,\n",
    "                      n_jobs=-1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Fitting 3 folds for each of 25 candidates, totalling 75 fits\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Parallel(n_jobs=-1)]: Using backend DaskDistributedBackend with 16 concurrent workers.\n",
      "[Parallel(n_jobs=-1)]: Done  18 tasks      | elapsed:    9.8s\n",
      "[Parallel(n_jobs=-1)]: Done  75 out of  75 | elapsed:   20.4s finished\n"
     ]
    }
   ],
   "source": [
    "# ### Fit Model with Dask\n",
    "from joblib import Parallel, parallel_backend\n",
    "with parallel_backend('dask'):\n",
    "  CV_rfc.fit(X_train, y_train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'max_depth': 2, 'n_estimators': 2}\n"
     ]
    }
   ],
   "source": [
    "# ### Show Best Parameters \n",
    "print(CV_rfc.best_params_)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Evaluate model performance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "              precision    recall  f1-score   support\n",
      "\n",
      "           0       1.00      1.00      1.00    144393\n",
      "           1       1.00      1.00      1.00      1783\n",
      "\n",
      "    accuracy                           1.00    146176\n",
      "   macro avg       1.00      1.00      1.00    146176\n",
      "weighted avg       1.00      1.00      1.00    146176\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from sklearn.metrics import classification_report\n",
    "# ## fit test dataset \n",
    "y_true, y_pred = y_test, CV_rfc.predict(X_test)\n",
    "# ## Print classification metrics \n",
    "print(classification_report(y_true, y_pred))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## print dask cluster config \n",
    "\n",
    "import json\n",
    "print(json.dumps(client.scheduler_info(), indent=4))"
   ]
  }
 ],
 "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}