{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "Scale Scikit-Learn for Small Data Problems\n", "==========================================\n", "\n", "This example demonstrates how Dask can scale scikit-learn to a cluster of machines for a CPU-bound problem.\n", "We'll fit a large model, a grid-search over many hyper-parameters, on a small dataset.\n", "\n", "This video talks demonstrates the same example on a larger cluster." ] }, { "cell_type": "code", "execution_count": 1, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:49:56.749276Z", "iopub.status.busy": "2021-10-14T16:49:56.748661Z", "iopub.status.idle": "2021-10-14T16:49:56.773204Z", "shell.execute_reply": "2021-10-14T16:49:56.774164Z" }, "keep_output": true }, "outputs": [ { "data": { "text/html": [ "\n", " \n", " " ], "text/plain": [ "" ] }, "execution_count": 1, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from IPython.display import IFrame\n", "\n", "IFrame(\"https://www.youtube.com/embed/5Zf6DQaf7jk\", 560, 315, frameborder=\"0\", allow=\"autoplay; encrypted-media\")" ] }, { "cell_type": "code", "execution_count": 2, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:49:56.777930Z", "iopub.status.busy": "2021-10-14T16:49:56.776604Z", "iopub.status.idle": "2021-10-14T16:50:00.081274Z", "shell.execute_reply": "2021-10-14T16:50:00.082080Z" } }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "/usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages/distributed/node.py:151: UserWarning: Port 8787 is already in use.\n", "Perhaps you already have a cluster running?\n", "Hosting the HTTP server on port 35455 instead\n", " warnings.warn(\n" ] }, { "data": { "text/html": [ "\n", "\n", "\n", "\n", "\n", "
\n", "

Client

\n", "\n", "		\n", "

Cluster

\n", "
    \n", "
  • Workers: 4
    • \n", "
  • Cores: 4
    • \n", "
  • Memory: 8.00 GB
    • \n", "
\n", "
" ], "text/plain": [ "" ] }, "execution_count": 2, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from dask.distributed import Client, progress\n", "client = Client(n_workers=4, threads_per_worker=1, memory_limit='2GB')\n", "client" ] }, { "cell_type": "markdown", "metadata": { "keep_output": true }, "source": [ "## Distributed Training\n", "\n", " \n", "\n", "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.\n", "\n", "Alternatively, Scikit-Learn can use Dask for parallelism. This lets you train those estimators using all the cores of your *cluster* without significantly changing your code.\n", "\n", "This is most useful for training large models on medium-sized datasets. You may have a large model when searching over many hyper-parameters, or when using an ensemble method with many individual estimators. For too small datasets, training times will typically be small enough that cluster-wide parallelism isn't helpful. For too large datasets (larger than a single machine's memory), the scikit-learn estimators may not be able to cope (though Dask-ML provides other ways for working with larger than memory datasets)." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Create Scikit-Learn Pipeline" ] }, { "cell_type": "code", "execution_count": 3, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:00.086157Z", "iopub.status.busy": "2021-10-14T16:50:00.085551Z", "iopub.status.idle": "2021-10-14T16:50:00.510314Z", "shell.execute_reply": "2021-10-14T16:50:00.511319Z" } }, "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": "code", "execution_count": 4, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:00.515253Z", "iopub.status.busy": "2021-10-14T16:50:00.514030Z", "iopub.status.idle": "2021-10-14T16:50:12.878572Z", "shell.execute_reply": "2021-10-14T16:50:12.879564Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Loading 20 newsgroups dataset for categories:\n", "['alt.atheism', 'talk.religion.misc']\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "857 documents\n", "2 categories\n", "\n" ] } ], "source": [ "# Scale Up: set categories=None to use all the categories\n", "categories = [\n", " 'alt.atheism',\n", " 'talk.religion.misc',\n", "]\n", "\n", "print(\"Loading 20 newsgroups dataset for categories:\")\n", "print(categories)\n", "\n", "data = fetch_20newsgroups(subset='train', categories=categories)\n", "print(\"%d documents\" % len(data.filenames))\n", "print(\"%d categories\" % len(data.target_names))\n", "print()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "We'll define a small pipeline that combines text feature extraction with a simple classifier." ] }, { "cell_type": "code", "execution_count": 5, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:12.883886Z", "iopub.status.busy": "2021-10-14T16:50:12.882668Z", "iopub.status.idle": "2021-10-14T16:50:12.888252Z", "shell.execute_reply": "2021-10-14T16:50:12.889938Z" } }, "outputs": [], "source": [ "pipeline = Pipeline([\n", " ('vect', HashingVectorizer()),\n", " ('tfidf', TfidfTransformer()),\n", " ('clf', SGDClassifier(max_iter=1000)),\n", "])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Define Grid for Parameter Search" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Grid search over some parameters." ] }, { "cell_type": "code", "execution_count": 6, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:12.893426Z", "iopub.status.busy": "2021-10-14T16:50:12.892150Z", "iopub.status.idle": "2021-10-14T16:50:12.897827Z", "shell.execute_reply": "2021-10-14T16:50:12.898562Z" } }, "outputs": [], "source": [ "parameters = {\n", " 'tfidf__use_idf': (True, False),\n", " 'tfidf__norm': ('l1', 'l2'),\n", " 'clf__alpha': (0.00001, 0.000001),\n", " # 'clf__penalty': ('l2', 'elasticnet'),\n", " # 'clf__n_iter': (10, 50, 80),\n", "}" ] }, { "cell_type": "code", "execution_count": 7, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:12.902141Z", "iopub.status.busy": "2021-10-14T16:50:12.901054Z", "iopub.status.idle": "2021-10-14T16:50:12.906187Z", "shell.execute_reply": "2021-10-14T16:50:12.907081Z" } }, "outputs": [], "source": [ "grid_search = GridSearchCV(pipeline, parameters, n_jobs=-1, verbose=1, cv=3, refit=False)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "To fit this normally, we would write\n", "\n", "\n", "```python\n", "grid_search.fit(data.data, data.target)\n", "```\n", "\n", "That would use the default joblib backend (multiple processes) for parallelism.\n", "To use the Dask distributed backend, which will use a cluster of machines to train the model, perform the fit in a `parallel_backend` context." ] }, { "cell_type": "code", "execution_count": 8, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:12.911006Z", "iopub.status.busy": "2021-10-14T16:50:12.909736Z", "iopub.status.idle": "2021-10-14T16:50:22.807079Z", "shell.execute_reply": "2021-10-14T16:50:22.806583Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Fitting 3 folds for each of 8 candidates, totalling 24 fits\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Parallel(n_jobs=-1)]: Using backend DaskDistributedBackend with 4 concurrent workers.\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Parallel(n_jobs=-1)]: Done 24 out of 24 | elapsed: 9.9s finished\n" ] } ], "source": [ "import joblib\n", "\n", "with joblib.parallel_backend('dask'):\n", " grid_search.fit(data.data, data.target)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "If you had your distributed dashboard open during that fit, you'll notice that each worker performs some of the fit tasks." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Parallel, Distributed Prediction\n", "\n", "Sometimes, you're train on a small dataset, but need to predict for a much larger batch of data.\n", "In this case, you'd like your estimator to handle NumPy arrays and pandas DataFrames for training, and dask arrays or DataFrames for prediction. [`dask_ml.wrappers.ParallelPostFit`](http://ml.dask.org/modules/generated/dask_ml.wrappers.ParallelPostFit.html#dask_ml.wrappers.ParallelPostFit) provides exactly that. It's a meta-estimator. It does nothing during training; the underlying estimator (probably a scikit-learn estimator) will probably be in-memory on a single machine. But tasks like `predict`, `score`, etc. are parallelized and distributed.\n", "\n", "Most of the time, using `ParallelPostFit` is as simple as wrapping the original estimator.\n", "When used inside a GridSearch, you'll need to update the keys of the parameters, just like with any meta-estimator.\n", "The only complication comes when using `ParallelPostFit` with another meta-estimator like `GridSearchCV`. In this case, you'll need to prefix your parameter names with `estimator__`." ] }, { "cell_type": "code", "execution_count": 9, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:22.812561Z", "iopub.status.busy": "2021-10-14T16:50:22.811024Z", "iopub.status.idle": "2021-10-14T16:50:23.127681Z", "shell.execute_reply": "2021-10-14T16:50:23.128759Z" } }, "outputs": [], "source": [ "from sklearn.datasets import load_digits\n", "from sklearn.svm import SVC\n", "from dask_ml.wrappers import ParallelPostFit" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "We'll load the small NumPy arrays for training." ] }, { "cell_type": "code", "execution_count": 10, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:23.132111Z", "iopub.status.busy": "2021-10-14T16:50:23.131023Z", "iopub.status.idle": "2021-10-14T16:50:23.223015Z", "shell.execute_reply": "2021-10-14T16:50:23.224249Z" } }, "outputs": [ { "data": { "text/plain": [ "(1797, 64)" ] }, "execution_count": 10, "metadata": {}, "output_type": "execute_result" } ], "source": [ "X, y = load_digits(return_X_y=True)\n", "X.shape" ] }, { "cell_type": "code", "execution_count": 11, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:23.228339Z", "iopub.status.busy": "2021-10-14T16:50:23.227134Z", "iopub.status.idle": "2021-10-14T16:50:23.232742Z", "shell.execute_reply": "2021-10-14T16:50:23.233645Z" } }, "outputs": [], "source": [ "svc = ParallelPostFit(SVC(random_state=0, gamma='scale'))\n", "\n", "param_grid = {\n", " # use estimator__param instead of param\n", " 'estimator__C': [0.01, 1.0, 10],\n", "}\n", "\n", "grid_search = GridSearchCV(svc, param_grid, cv=3)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "And fit as usual." ] }, { "cell_type": "code", "execution_count": 12, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:23.237567Z", "iopub.status.busy": "2021-10-14T16:50:23.236119Z", "iopub.status.idle": "2021-10-14T16:50:26.195842Z", "shell.execute_reply": "2021-10-14T16:50:26.195108Z" } }, "outputs": [ { "data": { "text/plain": [ "GridSearchCV(cv=3, estimator=ParallelPostFit(estimator=SVC(random_state=0)),\n", " param_grid={'estimator__C': [0.01, 1.0, 10]})" ] }, "execution_count": 12, "metadata": {}, "output_type": "execute_result" } ], "source": [ "grid_search.fit(X, y)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "We'll simulate a large dask array by replicating the training data a few times.\n", "In reality, you would load this from your file system." ] }, { "cell_type": "code", "execution_count": 13, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:26.201299Z", "iopub.status.busy": "2021-10-14T16:50:26.200473Z", "iopub.status.idle": "2021-10-14T16:50:26.209000Z", "shell.execute_reply": "2021-10-14T16:50:26.208516Z" } }, "outputs": [], "source": [ "import dask.array as da" ] }, { "cell_type": "code", "execution_count": 14, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:26.214927Z", "iopub.status.busy": "2021-10-14T16:50:26.214371Z", "iopub.status.idle": "2021-10-14T16:50:26.256850Z", "shell.execute_reply": "2021-10-14T16:50:26.257420Z" } }, "outputs": [ { "data": { "text/html": [ "\n", "\n", "\n", "\n", "\n", "
\n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
Array Chunk
Bytes 9.20 MB 920.06 kB
Shape (17970, 64) (1797, 64)
Count 11 Tasks 10 Chunks
Type float64 numpy.ndarray
\n", "		\n", "\n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "\n", " \n", " \n", " \n", "\n", " \n", " \n", "\n", " \n", " 64\n", " 17970\n", "\n", "
" ], "text/plain": [ "dask.array" ] }, "execution_count": 14, "metadata": {}, "output_type": "execute_result" } ], "source": [ "big_X = da.concatenate([\n", " da.from_array(X, chunks=X.shape)\n", " for _ in range(10)\n", "])\n", "big_X" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Operations like `predict`, or `predict_proba` return dask, rather than NumPy arrays.\n", "When you compute, the work will be done in parallel, out of core or distributed on the cluster." ] }, { "cell_type": "code", "execution_count": 15, "metadata": { "execution": { "iopub.execute_input": "2021-10-14T16:50:26.263700Z", "iopub.status.busy": "2021-10-14T16:50:26.263025Z", "iopub.status.idle": "2021-10-14T16:50:26.293159Z", "shell.execute_reply": "2021-10-14T16:50:26.293784Z" } }, "outputs": [ { "data": { "text/html": [ "\n", "\n", "\n", "\n", "\n", "
\n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
Array Chunk
Bytes 143.76 kB 14.38 kB
Shape (17970,) (1797,)
Count 21 Tasks 10 Chunks
Type int64 numpy.ndarray
\n", "		\n", "\n", "\n", " \n", " \n", " \n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "\n", " \n", " \n", "\n", " \n", " 17970\n", " 1\n", "\n", "
" ], "text/plain": [ "dask.array<_predict, shape=(17970,), dtype=int64, chunksize=(1797,), chunktype=numpy.ndarray>" ] }, "execution_count": 15, "metadata": {}, "output_type": "execute_result" } ], "source": [ "predicted = grid_search.predict(big_X)\n", "predicted" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "At this point predicted could be written to disk, or aggregated before returning to the client." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.8.12" } }, "nbformat": 4, "nbformat_minor": 2 }