dask-ml-scaling.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "eefe8bef-fba4-4006-a7aa-df78a49668f8",
   "metadata": {},
   "source": [
    "# Scaling Machine Learning Models Using Dask\n",
    "This tutorial demonstrates how we can scale a machine learning model in Dask.\n",
    "\n",
    "Learning outcomes of the tutorial are:\n",
    "1. Learn how to implement distributed training.\n",
    "2. Learn how to train for small dataset but predict for a much larger data.\n",
    "3. Learn how to incrementally train large datasets.\n",
    "4. Learn how to use Dask high-level collections to train on large datasets.\n",
    "\n",
    "Prerequisite:\n",
    "1. Experience with Scikit Learn library\n",
    "2. Experience with Dask Dataframe and Dask Arrays \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "73fc2081-80db-428e-88c0-647fc3945b82",
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.distributed as dd\n",
    "from dask.distributed import Client, LocalCluster, progress\n",
    "from dask_jobqueue import PBSCluster\n",
    "from dask.distributed import get_worker\n",
    "import dask.dataframe as dd\n",
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "98923f85-7bc5-4f94-b4cc-f84d6e54c212",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create the user directory if it doesn't already exist\n",
    "! mkdir -p /scratch/vp91/$USER"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "339f5d4b-8ccb-45e7-8559-8a16315930a7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# set the path \n",
    "user = os.getenv('USER', 'default value')\n",
    "path = '/scratch/vp91/'+user\n",
    "print(path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9da59131-ad46-4c3e-9a31-a32cba3e3356",
   "metadata": {},
   "outputs": [],
   "source": [
    "# The jupyter notebook is launched from your $HOME directory.\n",
    "# Change the working directory the user directory under /scratch/vp91\n",
    "os.chdir(os.path.expandvars(path))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fce81220-5ea8-4f22-be65-c75a58a7de4c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Make sure the python we use is from the venv\n",
    "os.environ['DASK_PYTHON'] = '/scratch/vp91/Training-Venv/intro-parallel-prog/bin/python3'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8f8c8a3c-3c8a-4d3d-9e07-2ea51cafb04c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Make sure all the modules are loaded.\n",
    "# It is essential that we use the same python and library for all aspects of dask\n",
    "# If we dont activate the venv then the workers may have a different versions of libraries\n",
    "setup_commands = [\"module load python3/3.11.0\", \"source /scratch/vp91/Training-Venv/intro-parallel-prog/bin/activate\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3a0e2dbb-4a9a-46a0-89d1-6076fbe0f2bf",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Gadi use custom PBS directives\n",
    "# So some of the default values to launch a PBS job through Dask call will not work in Gadi\n",
    "# Any directive specific to gadi should be mentioned here.\n",
    "# refer : https://opus.nci.org.au/display/Help/Gadi+Quick+Reference+Guide\n",
    "extra = ['-q normal',\n",
    "         '-P vp91', \n",
    "         '-l ncpus=48', \n",
    "         '-l mem=192GB']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1276d02c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# walltime: Walltime for each worker job.\n",
    "# cores: Total number of cores per job.\n",
    "# shebang: Path to desired interpreter for your batch submission script.\n",
    "# job_extra_directives: List of other PBS options. Each option will be prepended with the #PBS prefix.\n",
    "# local_directory: Dask worker local directory for file spilling.\n",
    "# job_directives_skip: Directives to skip in the generated job script header. Directives lines containing \n",
    "#                      the specified strings will be removed. Directives added by job_extra_directives \n",
    "#                      won’t be affected.\n",
    "# interface: Network interface like ‘eth0’ or ‘ib0’. This will be used both for the Dask scheduler and \n",
    "#            the Dask workers interface\n",
    "# job_script_prologue: Commands to add to script before launching worker\n",
    "# python: Python executable used to launch Dask workers. Defaults to the Python that is submitting these jobs\n",
    "\n",
    "\n",
    "\n",
    "cluster = PBSCluster(walltime=\"00:50:00\", \n",
    "                     cores=48, \n",
    "                     memory=\"192GB\",\n",
    "                     shebang='#!/usr/bin/env bash',\n",
    "                     job_extra_directives=extra, \n",
    "                     local_directory='$TMPDIR', \n",
    "                     job_directives_skip=[\"select\"], \n",
    "                     interface=\"ib0\",\n",
    "                     job_script_prologue=setup_commands,\n",
    "                     python=os.environ[\"DASK_PYTHON\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8dd5841c",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(cluster.job_script())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a1811896",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a cluster with 2 nodes\n",
    "cluster.scale(jobs=2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1dcf9b63-5a88-4958-b2df-3e9645a0a63a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Verify the workers have been allocated as expected\n",
    "!qstat"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b23e573c",
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3f91dd84",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create the client\n",
    "client = Client(cluster)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "76568e2a",
   "metadata": {},
   "outputs": [],
   "source": [
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ecb60d4e-492e-4138-b300-2af51b8277f0",
   "metadata": {},
   "source": [
    "## Distributed Training\n",
    "\n",
    "Distributed Training is particularly beneficial for training large models with medium-sized datasets. This scenario becomes relevant when dealing with extensive hyperparameter exploration or employing [ensemble method](https://scikit-learn.org/stable/modules/ensemble.html) involving numerous individual estimators."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c4ac8197-34c7-4eff-8aed-73ec286a766d",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pprint import pprint\n",
    "from time import time\n",
    "import logging\n",
    "\n",
    "from sklearn.datasets import fetch_20newsgroups\n",
    "from sklearn.feature_extraction.text import HashingVectorizer\n",
    "from sklearn.feature_extraction.text import TfidfTransformer\n",
    "from sklearn.linear_model import SGDClassifier\n",
    "from sklearn.model_selection import GridSearchCV\n",
    "from sklearn.pipeline import Pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "80924831-d692-4452-90fe-afeb62511010",
   "metadata": {},
   "source": [
    "To illustrate the concept of distributed training, we will utilize the [Newsgroup dataset]((https://scikit-learn.org/stable/modules/generated/sklearn.datasets.fetch_20newsgroups.html)) from Scikit-learn."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d2abba07-02f5-4c2c-b128-21d27c2a88d2",
   "metadata": {},
   "outputs": [],
   "source": [
    "data = fetch_20newsgroups(subset='train', categories=None)\n",
    "print(\"%d documents\" % len(data.filenames))\n",
    "print(\"%d categories\" % len(data.target_names))\n",
    "print()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "254c1bc7-4e40-47ae-bb3d-fa483b5955fe",
   "metadata": {},
   "source": [
    "### Create a pipeline:\n",
    "1. [HashingVectorizer](https://scikit-learn.org/stable/modules/generated/sklearn.feature_extraction.text.HashingVectorizer.html): Convert a collection of text documents to a matrix of token occurrences.\n",
    "2. [TfidfTransformer](https://scikit-learn.org/stable/modules/generated/sklearn.feature_extraction.text.TfidfTransformer.html): Transform a count matrix to a normalized tf or tf-idf representation.\n",
    "3. [SGDClassifier](https://scikit-learn.org/stable/modules/generated/sklearn.linear_model.SGDClassifier.html): This estimator implements regularized linear models with stochastic gradient descent (SGD) learning.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a1741453-0352-463e-a23c-1875728e2418",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline = Pipeline([\n",
    "    ('vect', HashingVectorizer()),\n",
    "    ('tfidf', TfidfTransformer()),\n",
    "    ('clf', SGDClassifier(max_iter=1000)),\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4e3b3173-19b5-4135-9e30-d1758cb3601b",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "515b1cec-81b2-435d-b09e-b8ddf55e8282",
   "metadata": {},
   "source": [
    "Each of these pipeline steps can possess distinct hyperparameters that significantly influence the model's accuracy. It is highly advisable to conduct a comprehensive search across a range of parameters within each step to identify the most suitable hyperparameters for the model. This process, known as hyperparameter tuning, is essential for optimizing the model's performance. In this tuturial we will be scoring a very small number of hyperparameters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7c6b5721-1629-43dd-a2ab-892cdf8881df",
   "metadata": {},
   "outputs": [],
   "source": [
    "parameters = {\n",
    "    'tfidf__use_idf': (True, False),\n",
    "    'tfidf__norm': ('l1', 'l2',  ),\n",
    "    'clf__alpha': (0.00001, 0.000001),\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b05af554-b368-4aa2-aa40-ccf08951e46c",
   "metadata": {},
   "source": [
    "In this tutorial, we leverage [GridSearchCV](https://scikit-learn.org/stable/modules/generated/sklearn.model_selection.GridSearchCV.html) to identify the most appropriate hyperparameters for the defined pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e2289916-eb76-440e-a844-48022e9bf9d2",
   "metadata": {},
   "outputs": [],
   "source": [
    "grid_search = GridSearchCV(pipeline, parameters, n_jobs=-1, verbose=1, cv=3, refit=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "27b0df16-87f2-4108-bcc2-59c82321bba5",
   "metadata": {},
   "source": [
    "Assessing hyperparameters with GridSearchCV involves a \"fit\" operation that demands substantial computational resources. To __fit__ this on a single node, we usually call the function\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "43a0c9eb-3d4d-430e-9161-f18ab88bf966",
   "metadata": {},
   "source": [
    "Scikit-learn uses [joblib](http://joblib.readthedocs.io/) for single-machine parallelism. This lets you train most estimators (anything that accepts an `n_jobs` parameter) using all the cores of your laptop or workstation. Alternatively, Scikit-Learn can use Dask for distributed parallelism.  This lets you train those estimators using all the cores of your *cluster* without significantly changing your code. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bb6b3831-2c1f-440e-8dc1-015e0f98594b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import joblib\n",
    "\n",
    "with joblib.parallel_backend('dask'):\n",
    "    grid_search.fit(data.data, data.target)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e7fa35fe-89e6-4bc6-8501-1273380a30cb",
   "metadata": {},
   "outputs": [],
   "source": [
    "grid_search.best_score_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3d7a0193-1478-4517-b856-ba36d2349891",
   "metadata": {},
   "outputs": [],
   "source": [
    "grid_search.best_params_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5b54adca-58e8-4a48-9e84-144ffac4b143",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "cv_results = pd.DataFrame(grid_search.cv_results_)\n",
    "cv_results.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "35e8dd62-b821-4729-8b96-caa33d371b39",
   "metadata": {},
   "source": [
    "## Score and Predict Large Datasets\n",
    "\n",
    "Sometimes you'll train on a smaller dataset that fits in memory, but need to predict or score for a much larger (possibly larger than memory) dataset. In this situation, you can use [ParallelPostFit](http://ml.dask.org/modules/generated/dask_ml.wrappers.ParallelPostFit.html) to parallelize and distribute the scoring or prediction steps. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4e348f57-78ff-4586-8832-f66ccfdf9a08",
   "metadata": {},
   "source": [
    "We'll generate a random dataset with scikit-learn to train the model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a80ede6c-8ebd-4d07-bf38-6d3421588427",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import dask.array as da\n",
    "from sklearn.datasets import make_classification"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2c5bbf39-74df-4ea1-8823-7b29d0e25585",
   "metadata": {},
   "source": [
    "We'll generate a random dataset with scikit-learn to train the model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a6e4f838-ad06-487f-887e-9742d7050f37",
   "metadata": {},
   "outputs": [],
   "source": [
    "X_train, y_train = make_classification(\n",
    "    n_features=2, n_redundant=0, n_informative=2,\n",
    "    random_state=1, n_clusters_per_class=1, n_samples=1000)\n",
    "X_train[:5]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c616f2d7-9df6-4c79-aac3-cbc507fc6666",
   "metadata": {},
   "source": [
    "_X_train_ and _y_train_ here is small enough to fit a on a single node. We will replicate this dataset multiple times with to create _X_large_ and _y_large_ and this represent the larger than memory dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5a69b1f7-3f58-4629-a0c2-c59a47d7b574",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Scale up: increase N, the number of times we replicate the data.\n",
    "N = 100\n",
    "X_large = da.concatenate([da.from_array(X_train, chunks=X_train.shape)\n",
    "                          for _ in range(N)])\n",
    "y_large = da.concatenate([da.from_array(y_train, chunks=y_train.shape)\n",
    "                          for _ in range(N)])\n",
    "X_large"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "15006569-d1d1-459e-af69-b6200836dfd9",
   "metadata": {},
   "source": [
    "Since our training dataset fits in memory, we can use a scikit-learn estimator as the actual estimator fit during traning. But we know that we’ll want to predict for a large dataset, so we’ll wrap the scikit-learn estimator with ParallelPostFit"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a2df439-4d93-400b-8d7d-3f7ad3aa89b2",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.linear_model import LogisticRegressionCV\n",
    "from dask_ml.wrappers import ParallelPostFit"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2d42be5b-13ff-44b3-ba6c-eb66ee2d3fba",
   "metadata": {},
   "outputs": [],
   "source": [
    "clf = ParallelPostFit(LogisticRegressionCV(cv=3), scoring=\"r2\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6f9c6473-e109-47d1-8b45-6f0c0c312e6c",
   "metadata": {},
   "source": [
    "See the note in the `dask-ml`'s documentation about when and why a `scoring` parameter is needed: https://ml.dask.org/modules/generated/dask_ml.wrappers.ParallelPostFit.html#dask_ml.wrappers.ParallelPostFit."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c01bd09b-e3e3-48bd-bfe6-76bb7ab6f6f6",
   "metadata": {},
   "source": [
    "Now we can train the model on the small dataset "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "314cc5d1-52ba-4998-95db-0d5239a924b1",
   "metadata": {},
   "outputs": [],
   "source": [
    "clf.fit(X_train, y_train)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc1cdf2f-9aef-4ca5-89fb-8c8ed1f211e5",
   "metadata": {},
   "source": [
    "Now that training is done, we'll turn to predicting for the large dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "54fe08d3-7354-4cfc-a7c4-fc132b78c38a",
   "metadata": {},
   "outputs": [],
   "source": [
    "y_pred = clf.predict(X_large)\n",
    "y_pred"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fef89dbf-60e2-4487-b6d7-3e9f6297d752",
   "metadata": {},
   "source": [
    "`y_pred` is a Dask array.\n",
    "Workers can write the predicted values to a shared file system, without ever having to collect the data on a single machine.\n",
    "\n",
    "Or we can check the models score on the entire large dataset.\n",
    "The computation will be done in parallel, and no single machine will have to hold all the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "408b0fe0-cc5d-40da-82b1-d0f628a7e9d4",
   "metadata": {},
   "outputs": [],
   "source": [
    "clf.score(X_large, y_large)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d9e195d0-bcbb-4e57-bc3c-b8b1015d9347",
   "metadata": {},
   "source": [
    "## Incremental Training\n",
    "\n",
    "When dealing with substantial datasets, it may become impractical to load the entire dataset into the computer's RAM simultaneously. Consequently, a more feasible approach involves loading the data in smaller, manageable chunks and training the model incrementally for each of these data subsets. Furthermore, in scenarios where fresh data continuously arrives over time, instead of retraining the model with the entire historical dataset, an incremental learning strategy can be employed. This approach preserves the prior knowledge of the model and allows for the incorporation of new data batches while maintaining the existing model's learning."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "316134f8-5823-4606-b5b0-d632e09487dd",
   "metadata": {},
   "source": [
    "Here we use a random dataset and split our dataset into training and testing data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "91a03757-1d32-434d-8829-10c75acfde47",
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask\n",
    "import dask.array as da\n",
    "from dask_ml.datasets import make_classification\n",
    "\n",
    "\n",
    "n, d = 100000, 100\n",
    "\n",
    "X, y = make_classification(n_samples=n, n_features=d,\n",
    "                           chunks=n // 10, flip_y=0.2)\n",
    "X"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dac5631b-0e8e-43c5-924a-771e2b448de1",
   "metadata": {},
   "source": [
    "We split our dataset into training and testing data to aid evaluation by making sure we have a fair test:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3d7053af-2acf-4164-ac58-999ddb66e91c",
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_ml.model_selection import train_test_split\n",
    "X_train, X_test, y_train, y_test = train_test_split(X, y)\n",
    "X_train"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0d6803af-39b7-4a6b-bfd9-ca588d2177ff",
   "metadata": {},
   "source": [
    "This dataset is small enough to fit in distributed memory, so we call `dask.persist` to ask Dask to execute the computations above and keep the results in memory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "867942b8-586b-46fc-b79e-ace486f03bae",
   "metadata": {},
   "outputs": [],
   "source": [
    "X_train, X_test, y_train, y_test = dask.persist(X_train, X_test, y_train, y_test)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d345ad3c-c721-441b-9804-4e00b7db1230",
   "metadata": {},
   "source": [
    "If you are working in a situation where your dataset does not fit in memory then you should skip this step.  Everything will still work, but will be slower and use less memory.\n",
    "\n",
    "Calling `dask.persist` will preserve our data in memory, so no computation will be needed as we pass over our data many times.  For example if our data came from CSV files and was not persisted, then the CSV files would have to be re-read on each pass.  This is desirable if the data does not fit in RAM, but not slows down our computation otherwise."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a9b04d00-81cc-4179-a81a-df9eb4a72dbd",
   "metadata": {},
   "source": [
    "We pre-compute the classes from our training data, which is required for this classification example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7688a4ce-8d9e-4859-aad2-560ef3b892c4",
   "metadata": {},
   "outputs": [],
   "source": [
    "classes = da.unique(y_train).compute()\n",
    "classes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b693a787-aa83-46fa-8eff-d533af5e5211",
   "metadata": {},
   "source": [
    "To incremental training we will use [SGDClassifier](https://scikit-learn.org/stable/modules/generated/sklearn.linear_model.SGDClassifier.html). But any estimator that implements the `partial_fit` method will work.  A list of Scikit-Learn models that implement this API is available [here](https://scikit-learn.org/stable/computing/scaling_strategies.html#incremental-learning)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f88b1253-9013-4307-9fd2-7df59e3607e3",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.linear_model import SGDClassifier\n",
    "\n",
    "est = SGDClassifier(loss='squared_error', penalty='l2', tol=1e-3)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "488723f4-4b17-4b51-aaa5-9ca73f5cbc2e",
   "metadata": {},
   "source": [
    "We now wrap our `SGDClassifer` with the [`dask_ml.wrappers.Incremental`](http://ml.dask.org/modules/generated/dask_ml.wrappers.Incremental.html#dask_ml.wrappers.Incremental) meta-estimator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "077fdb61-e7c8-4927-9da3-085bed03b0a2",
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_ml.wrappers import Incremental\n",
    "\n",
    "inc = Incremental(est, scoring='accuracy')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "049c90c1-ec2b-4058-9039-d6e14186a85c",
   "metadata": {},
   "source": [
    "`Incremental` only does data management while leaving the actual algorithm to the underlying Scikit-Learn estimator.\n",
    "Note: We set the scoring parameter above in the Dask estimator to tell it to handle scoring.  This works better when using Dask arrays for test data."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "33eca2d8-7e53-44a4-8961-83fd7d5edf04",
   "metadata": {},
   "source": [
    "`Incremental` implements a `fit` method, which will perform one loop over the dataset, calling `partial_fit` over each chunk in the Dask array. You may want to watch the dashboard during this fit process to see the sequential fitting of many batches."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fdf3cced-31f2-4e4b-9400-ea857f60a278",
   "metadata": {},
   "outputs": [],
   "source": [
    "inc.fit(X_train, y_train, classes=classes)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "47d7f538-fc76-41b6-a0f7-0eb55452fd51",
   "metadata": {},
   "outputs": [],
   "source": [
    "inc.score(X_test, y_test)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e59eafe2-7dcb-436a-8fe0-9349b5ff3c62",
   "metadata": {},
   "source": [
    "## Train Models on Large Datasets\n",
    "\n",
    "Estimators within scikit-learn are intended to operate with NumPy arrays or scipy sparse matrices and these data structures must be able to fit comfortably within the memory of a single machine. In contrast, estimators implemented in Dask-ML are optimized to effectively handle Dask Arrays and DataFrames. The advantage here is that Dask can manage much larger datasets compared to what can be accommodated in the memory of a single machine. These Dask-based data structures can be distributed across a cluster of machines, enabling efficient in-memory storage and processing of data on a much larger scale."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1bec82d1-c91a-4751-8660-b78cd3e2b4a6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask_ml.datasets\n",
    "import dask_ml.cluster\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "111f2071-d328-4c4a-9e20-e9ca8d872d38",
   "metadata": {},
   "source": [
    "In this example, we'll use `dask_ml.datasets.make_blobs` to generate some random *dask* arrays."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8ce5c2fd-8833-4c9d-99ce-b27355844bb5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Scale up: increase n_samples or n_features\n",
    "X, y = dask_ml.datasets.make_blobs(n_samples=1000000,\n",
    "                                   chunks=100000,\n",
    "                                   random_state=0,\n",
    "                                   centers=3)\n",
    "X = X.persist()\n",
    "X"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01aa7b9f-ba84-4a0e-a123-ec4b486b58e3",
   "metadata": {},
   "outputs": [],
   "source": [
    "X.visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a30a31d3-7147-4c99-b52c-f234eaf45e44",
   "metadata": {},
   "source": [
    "We'll use the k-means implemented in Dask-ML to cluster the points. It uses the `k-means||` (read: \"k-means parallel\") initialization algorithm, which scales better than `k-means++`. All of the computation, both during and after initialization, can be done in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c35794d0-5c7d-47d7-8478-1228b58e549d",
   "metadata": {},
   "outputs": [],
   "source": [
    "km = dask_ml.cluster.KMeans(n_clusters=3, init_max_iter=2, oversampling_factor=10)\n",
    "\n",
    "with joblib.parallel_backend('dask'):\n",
    "    km.fit(X)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e9317026-6376-4400-9a1e-4a0faa05ba9c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "fig, ax = plt.subplots()\n",
    "ax.scatter(X[::1000, 0], X[::1000, 1], marker='.', c=km.labels_[::1000],\n",
    "           cmap='viridis', alpha=0.25);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6a00fb80-2c42-42ea-a420-1e8162715d8d",
   "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.11.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}