3-machine-learning.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Parallel and Distributed Machine Learning\n",
    "\n",
    "*The material in this notebook is based on the open-source content from [Dask's tutorial repository](https://github.com/dask/dask-tutorial). Why? Because they did it so well. Thank you, Dask contributors!*\n",
    "\n",
    "We've now seen how Dask makes data analysis scalable with parallelization via Dask DataFrames. Let's now see how [Dask-ML](https://ml.dask.org/) allows us to do machine learning in a parallel and distributed manner. Note, machine learning is really just a special case of data analysis (one that automates analytical model building), so the 💪 Dask gains 💪 we've seen will apply here as well!\n",
    "\n",
    "(If you'd like a refresher on the difference between parallel and distributed computing, [here's a good discussion on StackExchange](https://cs.stackexchange.com/questions/1580/distributed-vs-parallel-computing).)\n",
    "\n",
    "In this notebook, we'll \n",
    "\n",
    "* break down machine learning scaling problems into two categories.\n",
    "* review how Scikit-Learn works.\n",
    "* solve an ML problem with a single machine (with Scikit-Learn).\n",
    "* solve an ML problem with a single machine and parallelism (with Scikit-Learn and Joblib).\n",
    "* solve an ML problem with a multiple machines and parallelism (with Scikit-Learn, Joblib, and Dask).\n",
    "* solve an ML problem with larger-than-memory data using multiple machines and parallelism (Dask-ML).\n",
    "* solve two ML problem with multiple machines *in the cloud* and parallelism (with Scikit-Learn, Joblib, Dask, and Coiled).\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Types of scaling problems in machine learning\n",
    "\n",
    "So you have your machine learning workflow that works well for small problems. Then there are two main types of scaling challenges you can run into: scaling the **size of your data** and scaling the **size of your model**. That is:\n",
    "\n",
    "1. **CPU-bound problems**: Data fits in RAM, but training takes too long. Many hyperparameter combinations, a large ensemble of many models, etc.\n",
    "2. **Memory-bound problems**: Data is larger than RAM, and sampling isn't an option.\n",
    "\n",
    "Here's a handy diagram for visualizing these problems:\n",
    "\n",
    "<img src=\"images/dimensions_of_scale.svg\" width=\"50%\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the bottom-left quadrant, your datasets aren’t too large (and therefore fit comfortably in RAM) and your model isn’t too large. Here, you’re much better off using something like scikit-learn, XGBoost, and similar libraries. You don't need to leverage multiple machines in a distributed manner with a library like Dask-ML here.\n",
    "\n",
    "If you’re in any of the other quadrants, however, distributed machine learning is the way to go.\n",
    "\n",
    "Here's a bird's eye view of the strategy we'll apply in this notebook:\n",
    "\n",
    "* For in-memory problems, just use scikit-learn (or your favorite ML library).\n",
    "* For large models, use `dask_ml.joblib` and your favorite scikit-learn estimator.\n",
    "* For large datasets, use `dask_ml` estimators."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Scikit-Learn in five minutes\n",
    "\n",
    "<img src=\"images/scikit_learn_logo_small.svg\" alt=\"scikit-learn logo\"/>\n",
    "\n",
    "\n",
    "In this section, we'll quickly run through a typical Scikit-Learn workflow:\n",
    "\n",
    "* Load some data (in this case, we'll generate it)\n",
    "* Import the Scikit-Learn module for our chosen ML algorithm\n",
    "* Create an estimator for that algorithm and fit it with our data\n",
    "* Inspect the learned attributes\n",
    "* Check the accurary of our model\n",
    "\n",
    "Scikit-Learn has a nice, consistent API:\n",
    "\n",
    "* You instantiate an `Estimator` (e.g. `LinearRegression`, `RandomForestClassifier`, etc.). All of the models *hyperparameters* (user-specified parameters, not the ones learned by the estimator) are passed to the estimator when it's created.\n",
    "* You call `estimator.fit(X, y)` to train the estimator.\n",
    "* Use `estimator` to inspect attributes, make predictions, etc. \n",
    "\n",
    "Here `X` is an array of *feature variables* (what you're using to predict) and `y` is an array of *target variables* (what we're trying to predict)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here's the workflow. Let's first generate some random data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[-0.77244139,  0.3607576 , -2.38110133,  0.08757   ],\n",
       "       [ 1.14946035,  0.62254594,  0.37302939,  0.45965795],\n",
       "       [-1.90879217, -1.1602627 , -0.27364545, -0.82766028],\n",
       "       [-0.77694695,  0.31434299, -2.26231851,  0.06339125],\n",
       "       [-1.17047054,  0.02212382, -2.17376797, -0.13421976],\n",
       "       [ 0.79010037,  0.68530624, -0.44740487,  0.44692959],\n",
       "       [ 1.68616989,  1.6329131 , -1.42072654,  1.04050557],\n",
       "       [-0.93912893, -1.02270838,  1.10093827, -0.63714432]])"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from sklearn.datasets import make_classification\n",
    "\n",
    "# Generate data\n",
    "X, y = make_classification(n_samples=10000, n_features=4, random_state=0)\n",
    "\n",
    "# View X\n",
    "X[:8]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*A quick [ML refresher](https://scikit-learn.org/stable/getting_started.html):*\n",
    "- *`X` is the samples matrix (or design matrix). The size of `X` is typically (`n_samples`, `n_features`), which means that samples are represented as rows and features are represented as columns.*\n",
    "- *A \"feature\" (also called an \"attribute\") is a measurable property of the phenomenon we're trying to analyze. A feature for a dataset of employees might be their hire date, for example.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([0, 0, 1, 0, 0, 0, 0, 1])"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# View y\n",
    "y[:8]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*A quick [ML refresher](https://scikit-learn.org/stable/getting_started.html):*\n",
    "- *`y` are the target values, which are real numbers for regression tasks, or integers for classification (or any other discrete set of values). For unsupervized learning tasks, `y` does not need to be specified. `y` is usually 1d array where the `i`th entry corresponds to the target of the `i`th sample (row) of `X`.*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll fit a [Support Vector Classifier](https://scikit-learn.org/stable/modules/generated/sklearn.svm.SVC.html) for this example, so let's load the appropriate Scikit-Learn module."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.svm import SVC"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we create the estimator and fit it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "SVC(random_state=0)"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "estimator = SVC(random_state=0)\n",
    "estimator.fit(X, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We inspect the learned features."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[-0.77244139,  0.3607576 , -2.38110133,  0.08757   ],\n",
       "       [ 1.14946035,  0.62254594,  0.37302939,  0.45965795],\n",
       "       [-0.77694695,  0.31434299, -2.26231851,  0.06339125],\n",
       "       [ 0.79010037,  0.68530624, -0.44740487,  0.44692959]])"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "estimator.support_vectors_[:4]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And check the accuracy."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.905"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "estimator.score(X, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*A quick ML refresher:*\n",
    "- *There are [3 different approaches](https://scikit-learn.org/0.15/modules/model_evaluation.html) to evaluate the quality of predictions of a model. One of them is the **estimator score method**. Estimators have a score method providing a default evaluation criterion for the problem they are designed to solve, which is discussed in each estimator's documentation.*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Hyperparameters"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "All estimators have parameters (often called *hyperparameters*). They affect the fit, but are specified up front instead of learned during training."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "SVC(C=1e-05, random_state=0, shrinking=False)"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "estimator = SVC(C=0.00001, shrinking=False, random_state=0)\n",
    "estimator.fit(X, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And the score of this hyperparameter set for this model:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.5007"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "estimator.score(X, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Hyperparameter Optimization\n",
    "\n",
    "There are a few ways to learn the best *hyper*parameters while training. One is `GridSearchCV`.\n",
    "As the name implies, this does a brute-force search over a grid of hyperparameter combinations.\n",
    "\n",
    "*A quick [ML refresher](https://scikit-learn.org/stable/getting_started.html):*\n",
    "- *Scikit-learn provides tools to automatically find the best parameter combinations via cross-validation (which is the \"CV\" in `GridSearchCV`).*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.model_selection import GridSearchCV"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Fitting 2 folds for each of 4 candidates, totalling 8 fits\n",
      "[CV] END ................................C=0.001, kernel=rbf; total time=   7.3s\n",
      "[CV] END ................................C=0.001, kernel=rbf; total time=   7.2s\n",
      "[CV] END ...............................C=0.001, kernel=poly; total time=   2.5s\n",
      "[CV] END ...............................C=0.001, kernel=poly; total time=   2.5s\n",
      "[CV] END .................................C=10.0, kernel=rbf; total time=   1.9s\n",
      "[CV] END .................................C=10.0, kernel=rbf; total time=   1.8s\n",
      "[CV] END ................................C=10.0, kernel=poly; total time=   2.4s\n",
      "[CV] END ................................C=10.0, kernel=poly; total time=   2.2s\n",
      "CPU times: user 33.4 s, sys: 230 ms, total: 33.6 s\n",
      "Wall time: 33.6 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "GridSearchCV(cv=2,\n",
       "             estimator=SVC(gamma='auto', probability=True, random_state=0),\n",
       "             param_grid={'C': [0.001, 10.0], 'kernel': ['rbf', 'poly']},\n",
       "             verbose=2)"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "estimator = SVC(gamma='auto', random_state=0, probability=True)\n",
    "param_grid = {\n",
    "    'C': [0.001, 10.0],\n",
    "    'kernel': ['rbf', 'poly'],\n",
    "}\n",
    "\n",
    "# Brute-force search over a grid of hyperparameter combinations\n",
    "grid_search = GridSearchCV(estimator, param_grid, verbose=2, cv=2)\n",
    "grid_search.fit(X, y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "({'C': 10.0, 'kernel': 'rbf'}, 0.9086000000000001)"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "grid_search.best_params_, grid_search.best_score_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Section recap:** We have\n",
    "\n",
    "* segmented ML scaling problems into two categories.\n",
    "  * CPU-bound problems where scaling up the model size is the issue\n",
    "  * RAM-bound problems where scaling up the data size is the issue\n",
    "* carried out a typical Scikit-Learn workflow for ML problems with small model(s) and small data.\n",
    "* reviewed hyperparameters and hyperparameter optimization in Scikit-Learn."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Compute Bound: Single-machine parallelism with Joblib\n",
    "\n",
    "<img src=\"images/joblib_logo.svg\" alt=\"Joblib logo\" style=\"width: 300px;\"/>\n",
    "\n",
    "In this section we'll see how [Joblib](https://joblib.readthedocs.io/en/latest/) (\"*a set of tools to provide lightweight pipelining in Python*\") gives us parallelism on our laptop. Here's what our grid search graph would look like if we set up six training \"jobs\" in parallel:\n",
    "\n",
    "![](images/unmerged_grid_search_graph.svg)\n",
    "\n",
    "With Joblib, we can say that Scikit-Learn has *single-machine* parallelism.\n",
    "Any Scikit-Learn estimator that can operate in parallel exposes an `n_jobs` keyword, which tells you how many tasks to run in parallel. Specifying `-1` jobs means running the maximum possible number of tasks in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Fitting 2 folds for each of 4 candidates, totalling 8 fits\n",
      "CPU times: user 5.72 s, sys: 182 ms, total: 5.91 s\n",
      "Wall time: 15.2 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "GridSearchCV(cv=2,\n",
       "             estimator=SVC(gamma='auto', probability=True, random_state=0),\n",
       "             n_jobs=-1,\n",
       "             param_grid={'C': [0.001, 10.0], 'kernel': ['rbf', 'poly']},\n",
       "             verbose=2)"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "grid_search = GridSearchCV(estimator, param_grid, verbose=2, cv=2, n_jobs=-1)\n",
    "grid_search.fit(X, y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "({'C': 10.0, 'kernel': 'rbf'}, 0.9086000000000001)"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "grid_search.best_params_, grid_search.best_score_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Section recap:** Previously, we\n",
    "\n",
    "* solved an ML problem with a single machine (with Scikit-Learn).\n",
    "\n",
    "In this section, we\n",
    "\n",
    "* solved an ML problem with a single machine and parallelism (with Scikit-Learn and Joblib)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Compute Bound: Multi-machine parallelism with Dask\n",
    "\n",
    "<img src=\"images/dask_horizontal_no_pad.svg\" alt=\"Dask logo\" style=\"width: 500px;\"/>\n",
    "\n",
    "In this section we'll see how Dask (plus Joblib and Scikit-Learn) gives us multi-machine parallelism. Here's what our grid search graph would look like if we allowed Dask to schedule our training \"jobs\" over multiple machines in our cluster:\n",
    "\n",
    "![](images/merged_grid_search_graph.svg)\n",
    "\n",
    "We can say that Dask can talk to Scikit-Learn (via Joblib) so that our *cluster* is used to train a model. \n",
    "\n",
    "If we run this on a laptop, it will take quite some time, but the CPU usage will be satisfyingly near 100% for the duration. To run faster, we would need a distributed cluster. That would mean putting something in the call to `Client` something like\n",
    "\n",
    "```\n",
    "c = Client('tcp://my.scheduler.address:8786')\n",
    "```\n",
    "\n",
    "Details on the many ways to create a cluster can be found [here](https://docs.dask.org/en/latest/setup/single-distributed.html)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's try it on a larger problem (more hyperparameters). First we'll instantiante a Dask Client."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "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-b70e1a4c-15d0-11ec-bb39-acde48001122</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> distributed.LocalCluster</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:8787/status\" target=\"_blank\">http://127.0.0.1:8787/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;\">LocalCluster</h3>\n",
       "        <p style=\"color: #9D9D9D; margin-bottom: 0px;\">07d4c5b1</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:8787/status\" target=\"_blank\">http://127.0.0.1:8787/status</a>\n",
       "                </td>\n",
       "                <td style=\"text-align: left;\">\n",
       "                    <strong>Workers:</strong> 4\n",
       "                </td>\n",
       "            </tr>\n",
       "            <tr>\n",
       "                <td style=\"text-align: left;\">\n",
       "                    <strong>Total threads:</strong> 12\n",
       "                </td>\n",
       "                <td style=\"text-align: left;\">\n",
       "                    <strong>Total memory:</strong> 16.00 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-7db1af54-67d4-4d91-9487-aa22f48c0178</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:55672\n",
       "                    </td>\n",
       "                    <td style=\"text-align: left;\">\n",
       "                        <strong>Workers:</strong> 4\n",
       "                    </td>\n",
       "                </tr>\n",
       "                <tr>\n",
       "                    <td style=\"text-align: left;\">\n",
       "                        <strong>Dashboard:</strong> <a href=\"http://127.0.0.1:8787/status\" target=\"_blank\">http://127.0.0.1:8787/status</a>\n",
       "                    </td>\n",
       "                    <td style=\"text-align: left;\">\n",
       "                        <strong>Total threads:</strong> 12\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> 16.00 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://192.168.1.14:55683\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Total threads: </strong> 3\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Dashboard: </strong> <a href=\"http://192.168.1.14:55686/status\" target=\"_blank\">http://192.168.1.14:55686/status</a>\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Memory: </strong> 4.00 GiB\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Nanny: </strong> tcp://127.0.0.1:55677\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> /Users/pavithra-coiled/Developer/Confs/data-science-at-scale/dask-worker-space/worker-rpb7ccig\n",
       "                        </td>\n",
       "                    </tr>\n",
       "\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://192.168.1.14:55685\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Total threads: </strong> 3\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Dashboard: </strong> <a href=\"http://192.168.1.14:55687/status\" target=\"_blank\">http://192.168.1.14:55687/status</a>\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Memory: </strong> 4.00 GiB\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Nanny: </strong> tcp://127.0.0.1:55674\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> /Users/pavithra-coiled/Developer/Confs/data-science-at-scale/dask-worker-space/worker-kgscmnhj\n",
       "                        </td>\n",
       "                    </tr>\n",
       "\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: 2</h4>\n",
       "                </summary>\n",
       "                <table style=\"width: 100%; text-align: left;\">\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Comm: </strong> tcp://192.168.1.14:55682\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Total threads: </strong> 3\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Dashboard: </strong> <a href=\"http://192.168.1.14:55689/status\" target=\"_blank\">http://192.168.1.14:55689/status</a>\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Memory: </strong> 4.00 GiB\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Nanny: </strong> tcp://127.0.0.1:55676\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> /Users/pavithra-coiled/Developer/Confs/data-science-at-scale/dask-worker-space/worker-q5373m7m\n",
       "                        </td>\n",
       "                    </tr>\n",
       "\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: 3</h4>\n",
       "                </summary>\n",
       "                <table style=\"width: 100%; text-align: left;\">\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Comm: </strong> tcp://192.168.1.14:55684\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Total threads: </strong> 3\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Dashboard: </strong> <a href=\"http://192.168.1.14:55688/status\" target=\"_blank\">http://192.168.1.14:55688/status</a>\n",
       "                        </td>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Memory: </strong> 4.00 GiB\n",
       "                        </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <td style=\"text-align: left;\">\n",
       "                            <strong>Nanny: </strong> tcp://127.0.0.1:55675\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> /Users/pavithra-coiled/Developer/Confs/data-science-at-scale/dask-worker-space/worker-nm8vsq8n\n",
       "                        </td>\n",
       "                    </tr>\n",
       "\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:55672' processes=4 threads=12, memory=16.00 GiB>"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import joblib\n",
    "import dask.distributed\n",
    "\n",
    "client = dask.distributed.Client()\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here we expand our problem by specifying more hyperparameters before training."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "param_grid = {\n",
    "    'C': [0.001, 0.1, 1.0, 2.5, 5, 10.0],\n",
    "    # Uncomment this for larger Grid searches on a cluster\n",
    "    'kernel': ['rbf', 'poly', 'linear'],\n",
    "    'shrinking': [True, False],\n",
    "}\n",
    "\n",
    "grid_search = GridSearchCV(estimator, param_grid, verbose=2, cv=2, n_jobs=-1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, watch this. We can fit our estimator with multi-machine parallelism by quickly *switching to a Dask parallel backend*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Fitting 2 folds for each of 36 candidates, totalling 72 fits\n",
      "CPU times: user 8.35 s, sys: 682 ms, total: 9.03 s\n",
      "Wall time: 48.9 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "with joblib.parallel_backend(\"dask\", scatter=[X, y]):\n",
    "    grid_search.fit(X, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "How does this work so seamlessly? Dask-ML developers worked with the Scikit-Learn and Joblib developers to implement a Dask parallel backend. So internally, scikit-learn now talks to Joblib, and Joblib talks to Dask, and Dask is what handles scheduling all of those tasks on multiple machines.\n",
    "\n",
    "The best parameters and best score:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "({'C': 10.0, 'kernel': 'rbf', 'shrinking': True}, 0.9086000000000001)"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "grid_search.best_params_, grid_search.best_score_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Section recap:** Previously, we\n",
    "\n",
    "* solved an ML problem with a single machine (with Scikit-Learn).\n",
    "* solved an ML problem with a single machine and parallelism (with Scikit-Learn and Joblib).\n",
    "\n",
    "In this section, we\n",
    "\n",
    "* solved an ML problem with multiple machines and parallelism (with Scikit-Learn, Joblib and Dask)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "## 3. Memory Bound: Single/Multi machine parallelism with Dask-ML\n",
    "\n",
    "We have seen how to work with larger models, but sometimes you'll want to train on a larger than memory dataset. `dask-ml` has implemented estimators that work well on Dask Arrays and DataFrames that may be larger than your machine's RAM."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.array as da\n",
    "import dask.delayed\n",
    "from sklearn.datasets import make_blobs\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll make a small (random) dataset locally using Scikit-Learn."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[ 1.00796679,  4.34582168,  2.15175661,  1.04337835, -1.82115164,\n",
       "         2.81149666, -1.18757701,  7.74628882,  9.36761449, -2.20570731,\n",
       "         5.71142324,  0.41084221,  1.34168817,  8.4568751 , -8.59042755,\n",
       "        -8.35194302, -9.55383028,  6.68605157,  5.34481483,  7.35044606],\n",
       "       [ 9.49283024,  6.1422784 , -0.97484846,  5.8604399 , -7.61126963,\n",
       "         2.86555735, -7.25390288,  8.89609285,  0.33510318, -1.79181328,\n",
       "        -4.66192239,  5.43323887, -0.86162507,  1.3705568 , -9.7904172 ,\n",
       "         2.3613231 ,  2.20516237,  2.20604823,  8.76464833,  3.47795068],\n",
       "       [-2.67206588, -1.30103177,  3.98418492, -8.88040428,  3.27735964,\n",
       "         3.51616445, -5.81395151, -7.42287114, -3.73476887, -2.89520363,\n",
       "         1.49435043, -1.35811028,  9.91250767, -7.86133474, -5.78975793,\n",
       "        -6.54897163,  3.08083281, -5.18975209, -0.85563107, -5.06615534],\n",
       "       [-6.85980599, -7.87144648,  3.33572279, -7.00394241, -5.97224874,\n",
       "        -2.55638942,  6.36329802, -7.97988653,  6.80059611, -8.14552537,\n",
       "         9.48255539, -0.67232114,  9.38462699,  2.09067352,  4.80505419,\n",
       "        -9.14866204, -4.32240399, -7.61670696, -4.14166466, -7.73998277]])"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "n_centers = 12\n",
    "n_features = 20\n",
    "\n",
    "X_small, y_small = make_blobs(n_samples=1000, centers=n_centers, n_features=n_features, random_state=0)\n",
    "\n",
    "centers = np.zeros((n_centers, n_features))\n",
    "\n",
    "for i in range(n_centers):\n",
    "    centers[i] = X_small[y_small == i].mean(0)\n",
    "    \n",
    "centers[:4]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The small dataset will be the template for our large random dataset.\n",
    "We'll use `dask.delayed` to adapt `sklearn.datasets.make_blobs`, so that the actual dataset is being generated on our workers. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "    <tr>\n",
       "        <td>\n",
       "            <table>\n",
       "                <thead>\n",
       "                    <tr>\n",
       "                        <td> </td>\n",
       "                        <th> Array </th>\n",
       "                        <th> Chunk </th>\n",
       "                    </tr>\n",
       "                </thead>\n",
       "                <tbody>\n",
       "                    \n",
       "                    <tr>\n",
       "                        <th> Bytes </th>\n",
       "                        <td> 14.90 GiB </td>\n",
       "                        <td> 30.52 MiB </td>\n",
       "                    </tr>\n",
       "                    \n",
       "                    <tr>\n",
       "                        <th> Shape </th>\n",
       "                        <td> (100000000, 20) </td>\n",
       "                        <td> (200000, 20) </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <th> Count </th>\n",
       "                        <td> 2000 Tasks </td>\n",
       "                        <td> 500 Chunks </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                    <th> Type </th>\n",
       "                    <td> float64 </td>\n",
       "                    <td> numpy.ndarray </td>\n",
       "                    </tr>\n",
       "                </tbody>\n",
       "            </table>\n",
       "        </td>\n",
       "        <td>\n",
       "        <svg width=\"75\" height=\"170\" style=\"stroke:rgb(0,0,0);stroke-width:1\" >\n",
       "\n",
       "  <!-- Horizontal lines -->\n",
       "  <line x1=\"0\" y1=\"0\" x2=\"25\" y2=\"0\" style=\"stroke-width:2\" />\n",
       "  <line x1=\"0\" y1=\"6\" x2=\"25\" y2=\"6\" />\n",
       "  <line x1=\"0\" y1=\"12\" x2=\"25\" y2=\"12\" />\n",
       "  <line x1=\"0\" y1=\"18\" x2=\"25\" y2=\"18\" />\n",
       "  <line x1=\"0\" y1=\"25\" x2=\"25\" y2=\"25\" />\n",
       "  <line x1=\"0\" y1=\"31\" x2=\"25\" y2=\"31\" />\n",
       "  <line x1=\"0\" y1=\"37\" x2=\"25\" y2=\"37\" />\n",
       "  <line x1=\"0\" y1=\"44\" x2=\"25\" y2=\"44\" />\n",
       "  <line x1=\"0\" y1=\"50\" x2=\"25\" y2=\"50\" />\n",
       "  <line x1=\"0\" y1=\"56\" x2=\"25\" y2=\"56\" />\n",
       "  <line x1=\"0\" y1=\"63\" x2=\"25\" y2=\"63\" />\n",
       "  <line x1=\"0\" y1=\"69\" x2=\"25\" y2=\"69\" />\n",
       "  <line x1=\"0\" y1=\"75\" x2=\"25\" y2=\"75\" />\n",
       "  <line x1=\"0\" y1=\"82\" x2=\"25\" y2=\"82\" />\n",
       "  <line x1=\"0\" y1=\"88\" x2=\"25\" y2=\"88\" />\n",
       "  <line x1=\"0\" y1=\"94\" x2=\"25\" y2=\"94\" />\n",
       "  <line x1=\"0\" y1=\"101\" x2=\"25\" y2=\"101\" />\n",
       "  <line x1=\"0\" y1=\"107\" x2=\"25\" y2=\"107\" />\n",
       "  <line x1=\"0\" y1=\"113\" x2=\"25\" y2=\"113\" />\n",
       "  <line x1=\"0\" y1=\"120\" x2=\"25\" y2=\"120\" style=\"stroke-width:2\" />\n",
       "\n",
       "  <!-- Vertical lines -->\n",
       "  <line x1=\"0\" y1=\"0\" x2=\"0\" y2=\"120\" style=\"stroke-width:2\" />\n",
       "  <line x1=\"25\" y1=\"0\" x2=\"25\" y2=\"120\" style=\"stroke-width:2\" />\n",
       "\n",
       "  <!-- Colored Rectangle -->\n",
       "  <polygon points=\"0.0,0.0 25.412616514582485,0.0 25.412616514582485,120.0 0.0,120.0\" style=\"fill:#8B4903A0;stroke-width:0\"/>\n",
       "\n",
       "  <!-- Text -->\n",
       "  <text x=\"12.706308\" y=\"140.000000\" font-size=\"1.0rem\" font-weight=\"100\" text-anchor=\"middle\" >20</text>\n",
       "  <text x=\"45.412617\" y=\"60.000000\" font-size=\"1.0rem\" font-weight=\"100\" text-anchor=\"middle\" transform=\"rotate(-90,45.412617,60.000000)\">100000000</text>\n",
       "</svg>\n",
       "        </td>\n",
       "    </tr>\n",
       "</table>"
      ],
      "text/plain": [
       "dask.array<concatenate, shape=(100000000, 20), dtype=float64, chunksize=(200000, 20), chunktype=numpy.ndarray>"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "n_samples_per_block = 200000\n",
    "n_blocks = 500\n",
    "\n",
    "delayeds = [dask.delayed(make_blobs)(n_samples=n_samples_per_block,\n",
    "                                     centers=centers,\n",
    "                                     n_features=n_features,\n",
    "                                     random_state=i)[0]\n",
    "            for i in range(n_blocks)]\n",
    "arrays = [da.from_delayed(obj, shape=(n_samples_per_block, n_features), dtype=X.dtype)\n",
    "          for obj in delayeds]\n",
    "X = da.concatenate(arrays)\n",
    "X"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "X = X.persist()  # Only run this on the cluster."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The algorithms implemented in Dask-ML are scalable. They handle larger-than-memory datasets just fine.\n",
    "\n",
    "They follow the scikit-learn API, so if you're familiar with Scikit-Learn, you'll feel at home with Dask-ML."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_ml.cluster import KMeans"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "clf = KMeans(init_max_iter=3, oversampling_factor=10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[CV] END .................................C=10.0, kernel=rbf; total time=   2.5s\n",
      "[CV] END .................................C=10.0, kernel=rbf; total time=   2.6s\n",
      "[CV] END ................................C=10.0, kernel=poly; total time=   3.0s\n",
      "[CV] END ................................C=10.0, kernel=poly; total time=   3.1s\n",
      "[CV] END ...............................C=0.001, kernel=poly; total time=   3.4s\n",
      "[CV] END ...............................C=0.001, kernel=poly; total time=   3.5s\n",
      "[CV] END ................................C=0.001, kernel=rbf; total time=   8.1s\n",
      "[CV] END ................................C=0.001, kernel=rbf; total time=   8.2s\n",
      "CPU times: user 1min 29s, sys: 21.6 s, total: 1min 51s\n",
      "Wall time: 7min 32s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "KMeans(init_max_iter=3, oversampling_factor=10)"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%time clf.fit(X)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "    <tr>\n",
       "        <td>\n",
       "            <table>\n",
       "                <thead>\n",
       "                    <tr>\n",
       "                        <td> </td>\n",
       "                        <th> Array </th>\n",
       "                        <th> Chunk </th>\n",
       "                    </tr>\n",
       "                </thead>\n",
       "                <tbody>\n",
       "                    \n",
       "                    <tr>\n",
       "                        <th> Bytes </th>\n",
       "                        <td> 381.47 MiB </td>\n",
       "                        <td> 781.25 kiB </td>\n",
       "                    </tr>\n",
       "                    \n",
       "                    <tr>\n",
       "                        <th> Shape </th>\n",
       "                        <td> (100000000,) </td>\n",
       "                        <td> (200000,) </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                        <th> Count </th>\n",
       "                        <td> 3000 Tasks </td>\n",
       "                        <td> 500 Chunks </td>\n",
       "                    </tr>\n",
       "                    <tr>\n",
       "                    <th> Type </th>\n",
       "                    <td> int32 </td>\n",
       "                    <td> numpy.ndarray </td>\n",
       "                    </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=\"6\" y1=\"0\" x2=\"6\" y2=\"25\" />\n",
       "  <line x1=\"12\" y1=\"0\" x2=\"12\" y2=\"25\" />\n",
       "  <line x1=\"18\" y1=\"0\" x2=\"18\" y2=\"25\" />\n",
       "  <line x1=\"25\" y1=\"0\" x2=\"25\" y2=\"25\" />\n",
       "  <line x1=\"31\" y1=\"0\" x2=\"31\" y2=\"25\" />\n",
       "  <line x1=\"37\" y1=\"0\" x2=\"37\" y2=\"25\" />\n",
       "  <line x1=\"44\" y1=\"0\" x2=\"44\" y2=\"25\" />\n",
       "  <line x1=\"50\" y1=\"0\" x2=\"50\" y2=\"25\" />\n",
       "  <line x1=\"56\" y1=\"0\" x2=\"56\" y2=\"25\" />\n",
       "  <line x1=\"63\" y1=\"0\" x2=\"63\" y2=\"25\" />\n",
       "  <line x1=\"69\" y1=\"0\" x2=\"69\" y2=\"25\" />\n",
       "  <line x1=\"75\" y1=\"0\" x2=\"75\" y2=\"25\" />\n",
       "  <line x1=\"82\" y1=\"0\" x2=\"82\" y2=\"25\" />\n",
       "  <line x1=\"88\" y1=\"0\" x2=\"88\" y2=\"25\" />\n",
       "  <line x1=\"94\" y1=\"0\" x2=\"94\" y2=\"25\" />\n",
       "  <line x1=\"101\" y1=\"0\" x2=\"101\" y2=\"25\" />\n",
       "  <line x1=\"107\" y1=\"0\" x2=\"107\" y2=\"25\" />\n",
       "  <line x1=\"113\" y1=\"0\" x2=\"113\" 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:#8B4903A0;stroke-width:0\"/>\n",
       "\n",
       "  <!-- Text -->\n",
       "  <text x=\"60.000000\" y=\"45.412617\" font-size=\"1.0rem\" font-weight=\"100\" text-anchor=\"middle\" >100000000</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<astype, shape=(100000000,), dtype=int32, chunksize=(200000,), chunktype=numpy.ndarray>"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "clf.labels_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*A quick [ML refresher](https://datascience.stackexchange.com/questions/44108/difference-between-a-target-and-a-label-in-machine-learning):*\n",
    "- *A label is a true outcome of the target variable `y`. In supervised learning, the target labels are known for the trainining dataset but not for the test.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([2, 1, 2, 0, 0, 0, 5, 1, 1, 3], dtype=int32)"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "clf.labels_[:10].compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Section recap:** Previously, we\n",
    "\n",
    "* solved an ML problem with a single machine (with Scikit-Learn).\n",
    "* solved an ML problem with a single machine and parallelism (with Scikit-Learn and Joblib).\n",
    "* solved an ML problem with multiple machines and parallelism (with Scikit-Learn, Joblib and Dask).\n",
    "\n",
    "In this section, we\n",
    "\n",
    "* solved an ML problem with larger-than-memory data using multiple machines and parallelism (Dask-ML)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Multi-machine parallelism in the cloud with Coiled\n",
    "\n",
    "<br>\n",
    "<img src=\"images/Coiled-Logo_Horizontal_RGB_Black.png\" alt=\"Coiled logo\" width=25%/>\n",
    "<br>\n",
    "\n",
    "In this section we'll see how Coiled allows us to solve machine learning problems with multi-machine parallelism in the cloud.\n",
    "\n",
    "Coiled, [among other things](https://coiled.io/product/), provides hosted and scalable Dask clusters. The biggest barriers to entry for doing machine learning at scale are \"Do you have access to a cluster?\" and \"Do you know how to manage it?\" Coiled solves both of those problems. Let's see how.\n",
    "\n",
    "We'll spin up a Coiled cluster (with 10 workers in this case), then instantiante a Dask Client to use with that cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import coiled\n",
    "from dask.distributed import Client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Spin up a Coiled cluster, instantiate a Client\n",
    "cluster = coiled.Cluster(n_workers=10, software=\"pavithraes/data-science-at-scale\",)\n",
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Compute bound: Dask parallel backend"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Again, we can fit our estimator with multi-machine paralellism by quickly switching to a Dask parallel backend. This time, this multi-machine parallelism is *in the cloud* because we've set up our Dask clusters via Coiled."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.datasets import make_classification\n",
    "\n",
    "X, y = make_classification(n_samples=10000, n_features=4, random_state=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "with joblib.parallel_backend(\"dask\", scatter=[X, y]):\n",
    "    grid_search.fit(X, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our cluster being a cloud-based cluster that adds no complexity is Coiled's mission on full display.\n",
    "\n",
    "The best parameters and best score:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "grid_search.best_params_, grid_search.best_score_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Memory bound: Dask-ML"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Similarly, we can use Dask-ML estimators on the cloud to work with larger datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_centers = 12\n",
    "n_features = 20\n",
    "\n",
    "X_small, y_small = make_blobs(n_samples=1000, centers=n_centers, n_features=n_features, random_state=0)\n",
    "\n",
    "centers = np.zeros((n_centers, n_features))\n",
    "\n",
    "for i in range(n_centers):\n",
    "    centers[i] = X_small[y_small == i].mean(0)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_samples_per_block = 200000\n",
    "n_blocks = 500\n",
    "\n",
    "delayeds = [dask.delayed(make_blobs)(n_samples=n_samples_per_block,\n",
    "                                     centers=centers,\n",
    "                                     n_features=n_features,\n",
    "                                     random_state=i)[0]\n",
    "            for i in range(n_blocks)]\n",
    "arrays = [da.from_delayed(obj, shape=(n_samples_per_block, n_features), dtype=X.dtype)\n",
    "          for obj in delayeds]\n",
    "X = da.concatenate(arrays)\n",
    "X = X.persist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_ml.cluster import KMeans"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clf = KMeans(init_max_iter=3, oversampling_factor=10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time clf.fit(X)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Computng the labels:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clf.labels_[:10].compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Section recap:** Previously, we\n",
    "\n",
    "* solved an ML problem with a single machine (with Scikit-Learn).\n",
    "* solved an ML problem with a single machine and parallelism (with Scikit-Learn and Joblib).\n",
    "* solved an ML problem with a multiple machines and parallelism (with Scikit-Learn, Joblib, and Dask).\n",
    "* solved an ML problem with larger-than-memory data using multiple machines and parallelism (Dask-ML).\n",
    "\n",
    "In this section, we\n",
    "\n",
    "* solved two ML problem with multiple machines *in the cloud* and parallelism (with Scikit-Learn, Joblib, Dask, and Coiled)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Recap (for this entire notebook!)\n",
    "\n",
    "We did the following:\n",
    "\n",
    "* broke down machine learning scaling problems into to two categories (data size vs. model size).\n",
    "* solved an ML problem with a single machine (with Scikit-Learn).\n",
    "* solved an ML problem with a single machine and parallelism (with Scikit-Learn and Joblib).\n",
    "* solved an ML problem with a multiple machines and parallelism (with Scikit-Learn, Joblib and Dask).\n",
    "* used `dask-ml` estimators that work well on Dask Arrays and DataFrames to train on datasets larger than your machine's RAM.\n",
    "* solved an ML problem with a multiple machines *in the cloud* and parallelism (with Scikit-Learn, Joblib, Dask and Coiled)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.10"
  },
  "nbsphinx": {
   "execute": "never"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}