06-ML.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask-ML\n",
    "====================================================================\n",
    "\n",
    "__(initial notes are courtesy of the Dask project homepage at ml.dask.org)__\n",
    "\n",
    "Dask-ML provides scalable machine learning in Python using [Dask](https://dask.org/) alongside popular machine learning libraries like [Scikit-Learn](http://scikit-learn.org/).\n",
    "\n",
    "The idea is to support Pandas + Scikit style ML for parallel scenarios, with code patterns you're used to:\n",
    "\n",
    "```python\n",
    "import dask.dataframe as dd\n",
    "df = dd.read_parquet('...')\n",
    "data = df[['age', 'income', 'married']]\n",
    "labels = df['outcome']\n",
    "\n",
    "from dask_ml.linear_model import LogisticRegression\n",
    "lr = LogisticRegression()\n",
    "lr.fit(data, labels)\n",
    "```\n",
    "\n",
    "How does this work?\n",
    "-------------------------------------------------------------------------------------------\n",
    "\n",
    "Modern machine learning algorithms employ a wide variety of techniques. Scaling these requires a similarly wide variety of different approaches. Generally solutions fall into the following three categories:\n",
    "\n",
    "### Parallelize Scikit-Learn Directly\n",
    "\n",
    "Scikit-Learn already provides parallel computing on a single machine with [Joblib](http://joblib.readthedocs.io/en/latest/). Dask extends this parallelism to many machines in a cluster. This works well for modest data sizes but large computations, such as random forests, hyper-parameter optimization, and more.\n",
    "\n",
    "```python\n",
    "from dask.distributed import Client\n",
    "import joblib\n",
    "\n",
    "client = Client()  # Connect to a Dask Cluster\n",
    "\n",
    "with joblib.parallel_backend('dask'):\n",
    "    # Your normal scikit-learn code here\n",
    "```\n",
    "\n",
    "See [Dask-ML Joblib documentation](https://ml.dask.org/joblib.html) for more information.\n",
    "\n",
    "*Note that this is an active collaboration with the Scikit-Learn development team. This functionality is progressing quickly but is in a state of rapid change.*\n",
    "\n",
    "### Reimplement Scalable Algorithms with Dask Array\n",
    "\n",
    "Some machine learning algorithms are easy to write down as Numpy algorithms. In these cases we can replace Numpy arrays with Dask arrays to achieve scalable algorithms easily. This is employed for [linear models](https://ml.dask.org/glm.html), [pre-processing](https://ml.dask.org/preprocessing.html), and [clustering](https://ml.dask.org/clustering.html).\n",
    "\n",
    "```python\n",
    "from dask_ml.preprocessing import Categorizer, DummyEncoder\n",
    "from dask_ml.linear_model import LogisticRegression\n",
    "\n",
    "lr = LogisticRegression()\n",
    "lr.fit(data, labels)\n",
    "```\n",
    "\n",
    "### Partner with other distributed libraries\n",
    "\n",
    "Other machine learning libraries like XGBoost and TensorFlow already have distributed solutions that work quite well. Dask-ML makes no attempt to re-implement these systems. Instead, Dask-ML makes it easy to use normal Dask workflows to prepare and set up data, then it deploys XGBoost or Tensorflow *alongside* Dask, and hands the data over.\n",
    "\n",
    "```python\n",
    "from dask_ml.xgboost import XGBRegressor\n",
    "\n",
    "est = XGBRegressor(...)\n",
    "est.fit(train, train_labels)\n",
    "```\n",
    "\n",
    "See [Dask-ML + XGBoost](https://ml.dask.org/xgboost.html) or [Dask-ML + TensorFlow](https://ml.dask.org/tensorflow.html) documentation for more information.\n",
    "\n",
    "Scikit-Learn API[](https://ml.dask.org/#scikit-learn-api \"Permalink to this headline\")\n",
    "--------------------------------------------------------------------------------------\n",
    "\n",
    "In all cases Dask-ML endeavors to provide a single unified interface around the familiar NumPy, Pandas, and Scikit-Learn APIs. Users familiar with Scikit-Learn should feel at home with Dask-ML.\n",
    "\n",
    "* * *\n",
    "\n",
    "# Let's try it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "\n",
    "client = Client(n_workers=2, threads_per_worker=1, memory_limit='1GB')\n",
    "\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.dataframe\n",
    "\n",
    "ddf = dask.dataframe.read_csv('data/diamonds.csv', blocksize=1e6, assume_missing=True)\n",
    "ddf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf = ddf.drop(['Unnamed: 0'], axis=1)\n",
    "ddf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf = ddf.categorize()\n",
    "ddf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "prepared = dask.dataframe.reshape.get_dummies(ddf)\n",
    "prepared"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Alternatively, we could use scikit-learn style preprocessing steps, though the relevant APIs are still evolving a bit:\n",
    "\n",
    "```python\n",
    "from sklearn.pipeline import make_pipeline\n",
    "from dask_ml.preprocessing import Categorizer, DummyEncoder\n",
    "from dask_ml.linear_model import LinearRegression\n",
    "\n",
    "pipe = make_pipeline(\n",
    "    Categorizer(),\n",
    "    DummyEncoder()\n",
    ")\n",
    "\n",
    "pipe.fit(ddf)\n",
    "\n",
    "prepared = pipe.transform(ddf)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "prepared.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "y = prepared.price.to_dask_array(lengths=True)\n",
    "y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "arr = prepared.drop('price', axis=1).to_dask_array(lengths=True)\n",
    "arr"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_ml.model_selection import train_test_split\n",
    "\n",
    "X_train, X_test, y_train, y_test = train_test_split(arr, y, test_size=0.1)\n",
    "\n",
    "X_train"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "y_train"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_ml.linear_model import LinearRegression\n",
    "\n",
    "lr = LinearRegression(solver='lbfgs', max_iter=10)\n",
    "lr_model = lr.fit(X_train, y_train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lr_model.coef_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "y_predicted = lr_model.predict(X_test)\n",
    "\n",
    "y_predicted"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "y_test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_ml.metrics import mean_squared_error\n",
    "from math import sqrt\n",
    "\n",
    "sqrt(mean_squared_error(y_test, y_predicted))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (Optional) Hands-On Lab\n",
    "\n",
    "We can take a quick try ourselves using the UC Irvine ML repository's *Combined Cycle Power Plant Data Set* (https://archive.ics.uci.edu/ml/datasets/Combined+Cycle+Power+Plant)\n",
    "\n",
    "This dataset consists of about 10,000 records of measurements relating to peaker power plants.\n",
    "- Temperature (AT) in the range 1.81°C and 37.11°C,\n",
    "- Ambient Pressure (AP) in the range 992.89-1033.30 millibar,\n",
    "- Relative Humidity (RH) in the range 25.56% to 100.16%\n",
    "- Exhaust Vacuum (V) in the range 25.36-81.56 cm Hg\n",
    "- Net hourly electrical energy output (PE) 420.26-495.76 MW\n",
    "\n",
    "We want to model the power output as a function of the other parameters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf = dask.dataframe.read_csv('data/powerplant.csv', sample=512000, blocksize=4e4)\n",
    "ddf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "(Insert more cells here as needed using the 'b' hotkey, the menu, or by copying and pasting from the earlier example)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.close()"
   ]
  }
 ],
 "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}