1-overview.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "<img src=\"images/dask_horizontal.svg\"\n",
    "     width=\"45%\"\n",
    "     alt=\"Dask logo\\\">\n",
    "     \n",
    "# Parallel Computing in Python with Dask\n",
    "\n",
    "This notebook provides a high-level overview of Dask. We discuss why you might want to use Dask, high-level and low-level APIs for generating computational graphs, and Dask's schedulers which enable the parallel execution of these graphs."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "# Overview\n",
    "\n",
    "[Dask](https://docs.dask.org) is a flexible, [open source](https://github.com/dask/dask) library for parallel and distributed computing in Python. Dask is designed to scale the existing Python ecosystem.\n",
    "\n",
    "You might want to use Dask because it:\n",
    "\n",
    "- Enables parallel and larger-than-memory computations\n",
    "\n",
    "- Uses familiar APIs you're used to from projects like NumPy, pandas, and scikit-learn\n",
    "\n",
    "- Allows you to scale existing workflows with minimal code changes\n",
    "\n",
    "- Dask works on your laptop, but also scales out to large clusters\n",
    "\n",
    "- Offers great built-in diagnosic tools"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Components of Dask\n",
    "\n",
    "From a high level, Dask is comprised of two main components:\n",
    "\n",
    "1. **Dask collections** which extend common interfaces like NumPy, pandas, and Python iterators to larger-than-memory or distributed environments by creating *task graphs*\n",
    "2. **Schedulers** which compute task graphs produced by Dask collections in parallel\n",
    "\n",
    "<img src=\"images/dask-overview.svg\"\n",
    "     width=\"85%\"\n",
    "     alt=\"Dask components\\\">"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task Graphs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def inc(i):\n",
    "    return i + 1\n",
    "\n",
    "def add(a, b):\n",
    "    return a + b\n",
    "\n",
    "a, b = 1, 12\n",
    "c = inc(a)\n",
    "d = inc(b)\n",
    "output = add(c, d)\n",
    "\n",
    "print(f'output = {output}')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This computation can be encoded in the following task graph:"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](images/inc-add.png)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "- Graph of inter-related tasks with dependencies between them\n",
    "\n",
    "- Circular nodes in the graph are Python function calls\n",
    "\n",
    "- Square nodes are Python objects that are created by one task as output and can be used as inputs in another task"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "# Dask Collections\n",
    "\n",
    "Let's looks at two Dask user interfaces: Dask Array and Dask Delayed.\n",
    "\n",
    "## Dask Arrays\n",
    "\n",
    "- Dask arrays are chunked, n-dimensional arrays\n",
    "\n",
    "- Can think of a Dask array as a collection of NumPy `ndarray` arrays\n",
    "\n",
    "- Dask arrays implement a large subset of the NumPy API using blocked algorithms\n",
    "\n",
    "- For many purposes Dask arrays can serve as drop-in replacements for NumPy arrays"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<img src=\"images/dask-array.png\" width=\"50%\">"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import dask.array as da"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "x_np = np.random.random(size=(1_000, 1_000))\n",
    "x_np"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can create a Dask array in a similar manner, but need to specify a `chunks` argument to tell Dask how to break up the underlying array into chunks."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "x = da.random.random(size=(1_000, 1_000), chunks=(250, 500))\n",
    "x  # Dask arrays have nice HTML output in Jupyter notebooks"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask arrays look and feel like NumPy arrays. For example, they have `dtype` and `shape` attributes:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(x.dtype)\n",
    "print(x.shape)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Dask collections are _lazily_ evaluated; the result from a computation isn't computed until you ask for it. Instead, a Dask task graph for the computation is produced. You can visualize this task graph using the `visualize()` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [],
   "source": [
    "x.visualize()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "To compute a task graph call the `compute()` method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "result = x.compute()    # We'll go into more detail about .compute() later on\n",
    "result"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The result of this computation is a familiar NumPy `ndarray`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(result)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask arrays support a large portion of the NumPy interface:\n",
    "\n",
    "- Arithmetic and scalar mathematics: `+`, `*`, `exp`, `log`, ...\n",
    "\n",
    "- Reductions along axes: `sum()`, `mean()`, `std()`, `sum(axis=0)`, ...\n",
    "\n",
    "- Tensor contractions / dot products / matrix multiply: `tensordot`\n",
    "\n",
    "- Axis reordering / transpose: `transpose`\n",
    "\n",
    "- Slicing: `x[:100, 500:100:-2]`\n",
    "\n",
    "- Fancy indexing along single axes with lists or numpy arrays: `x[:, [10, 1, 5]]`\n",
    "\n",
    "- Array protocols like `__array__` and `__array_ufunc__`\n",
    "\n",
    "- Some linear algebra: `svd`, `qr`, `solve`, `solve_triangular`, `lstsq`, ...\n",
    "\n",
    "- ...\n",
    "\n",
    "See the [Dask array API docs](http://docs.dask.org/en/latest/array-api.html) for full details about what portion of the NumPy API is implemented for Dask arrays."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can build more complex computations using the familiar NumPy operations we're used to."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "result = (x + x.T).sum(axis=0).mean()\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "result.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "result.compute()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Note**: Dask can be used to scale other array-like libraries that support the NumPy `ndarray` interface. For example, [pydata/sparse](https://sparse.pydata.org/en/latest/) for sparse arrays or [CuPy](https://cupy.dev/) for GPU-accelerated arrays."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dask Delayed\n",
    "\n",
    "Sometimes problems don’t fit nicely into one of the high-level collections like Dask arrays or Dask DataFrames. In these cases, you can parallelize custom algorithms using the lower-level Dask `delayed` interface. This allows one to manually create task graphs with a light annotation of normal Python code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "import random\n",
    "\n",
    "def inc(x):\n",
    "    time.sleep(random.random())\n",
    "    return x + 1\n",
    "\n",
    "def double(x):\n",
    "    time.sleep(random.random())\n",
    "    return 2 * x\n",
    "    \n",
    "def add(x, y):\n",
    "    time.sleep(random.random())\n",
    "    return x + y "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "data = [1, 2, 3, 4]\n",
    "\n",
    "output = []\n",
    "for i in data:\n",
    "    a = inc(i)\n",
    "    b = double(i)\n",
    "    c = add(a, b)\n",
    "    output.append(c)\n",
    "\n",
    "total = sum(output)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask `delayed` wraps function calls and delays their execution. `delayed` functions record what we want to compute (a function and input parameters) as a task in a graph that we’ll run later on parallel hardware by calling `compute`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask import delayed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@delayed\n",
    "def lazy_inc(x):\n",
    "    time.sleep(random.random())\n",
    "    return x + 1\n",
    "\n",
    "lazy_inc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inc_output = lazy_inc(3)  # lazily evaluate inc(3)\n",
    "inc_output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inc_output.compute()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using `delayed` functions, we can build up a task graph for the particular computation we want to perform"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "double_inc_output = lazy_inc(inc_output)\n",
    "double_inc_output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "double_inc_output.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "double_inc_output.compute()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can use `delayed` to make our previous example computation lazy by wrapping all the function calls with delayed:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "import random\n",
    "\n",
    "@delayed\n",
    "def inc(x):\n",
    "    time.sleep(random.random())\n",
    "    return x + 1\n",
    "\n",
    "@delayed\n",
    "def double(x):\n",
    "    time.sleep(random.random())\n",
    "    return 2 * x\n",
    "\n",
    "@delayed\n",
    "def add(x, y):\n",
    "    time.sleep(random.random())\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "data = [1, 2, 3, 4]\n",
    "\n",
    "output = []\n",
    "for i in data:\n",
    "    a = inc(i)\n",
    "    b = double(i)\n",
    "    c = add(a, b)\n",
    "    output.append(c)\n",
    "\n",
    "total = delayed(sum)(output)\n",
    "total"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "total.compute()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We highly recommend checking out the [Dask delayed best practices](http://docs.dask.org/en/latest/delayed-best-practices.html) page to avoid some common pitfalls when using `delayed`. "
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Schedulers\n",
    "\n",
    "High-level collections like Dask arrays and Dask DataFrames, as well as the low-level `dask.delayed` interface build up task graphs for a computation. After these graphs are generated, they need to be executed (potentially in parallel). This is the job of a task scheduler. Different task schedulers exist within Dask. Each will consume a task graph and compute the same result, but with different performance characteristics. "
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![grid-search](images/animation.gif \"grid-search\")\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask has two different classes of schedulers: \n",
    "\n",
    "1. **Single machine schedulers** provide basic features on a local process or thread pool with minimal setup.\n",
    "2. **Distributed schedulers** are more sophisticated and offer more features, but also require a bit more effort to set up."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Single Machine Schedulers\n",
    "\n",
    "Single machine schedulers provide basic features on a local process or thread pool and require no setup (only use the Python standard library). The different single machine schedulers Dask provides are:\n",
    "\n",
    "- `'threads'`: The threaded scheduler executes computations with a local `concurrent.futures.ThreadPoolExecutor`. The threaded scheduler is the default choice for Dask arrays, Dask DataFrames, and Dask delayed. \n",
    "\n",
    "- `'processes'`: The multiprocessing scheduler executes computations with a local `concurrent.futures.ProcessPoolExecutor`.\n",
    "\n",
    "- `'synchronous'`: The single-threaded synchronous scheduler executes all computations in the local thread, with no parallelism at all. This is particularly valuable for debugging and profiling, which are more difficult when using threads or processes."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can configure which scheduler is used in a few different ways. You can set the scheduler globally by using the `dask.config.set(scheduler=)` command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask\n",
    "\n",
    "dask.config.set(scheduler='threads')\n",
    "x.compute()   # Will use the multi-threading scheduler"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "or use it as a context manager to set the scheduler for a block of code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with dask.config.set(scheduler='processes'):\n",
    "    x.compute()  # Will use the multi-processing scheduler"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "or even within a single compute call:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "x.compute(scheduler='threads')  # Will use the multi-threading scheduler"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `num_workers` argument can be used to specify the number of threads or processes to use:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "x.compute(scheduler='threads', num_workers=4)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Distributed Scheduler\n",
    "\n",
    "Despite having \"distributed\" in its name, the distributed scheduler works well on both single and multiple machines. Think of it as an \"advanced\" scheduler.\n",
    "\n",
    "A Dask distributed cluster is composed of a single centralized scheduler and one or more worker processes. A `Client` object is used as the user-facing entry point to interact with the cluster. We will talk about the components of Dask clusters in more detail later on in [3-distributed-scheduler.ipynb](3-distributed-scheduler.ipynb).\n",
    "\n",
    "<img src=\"images/distributed-overview.svg\"\n",
    "     width=\"85%\"\n",
    "     alt=\"Dask components\\\">"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The distributed scheduler has many features:\n",
    "\n",
    "- [Real-time, `concurrent.futures`-like interface](https://docs.dask.org/en/latest/futures.html)\n",
    "\n",
    "- [Sophisticated memory management](https://distributed.dask.org/en/latest/memory.html)\n",
    "\n",
    "- [Data locality](https://distributed.dask.org/en/latest/locality.html)\n",
    "\n",
    "- [Adaptive deployments](https://distributed.dask.org/en/latest/adaptive.html)\n",
    "\n",
    "- [Cluster resilience](https://distributed.dask.org/en/latest/resilience.html)\n",
    "\n",
    "- ...\n",
    "\n",
    "See the [Dask distributed documentation](https://distributed.dask.org) for full details about all the distributed scheduler features."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "\n",
    "# Creates a local Dask cluster\n",
    "client = Client()\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "x = da.ones((20_000, 20_000), chunks=(400, 400))\n",
    "result = (x + x.T).sum(axis=0).mean()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "result.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.close()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Next steps\n",
    "\n",
    "Next, let's learn more about Dask internals and how to do custom operations in the [2-custom-operations.ipynb](2-custom-operations.ipynb) notebook."
   ]
  }
 ],
 "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.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}