002-intro-to-dask.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction to Dask\n",
    "\n",
    "[Dask](https://dask.org/) is a library for scalable computing in Python. It primarily does two things: \n",
    "\n",
    "1. Scheduling tasks on a cluster. This helps parallelize intensive computational tasks. \n",
    "\n",
    "2. Contains implementations of collections for large datasets. This helps manage large datasets, so that scheduling tasks on them becomes more easy to do.\n",
    "\n",
    "Let's start with scheduling and how that works. We'll start with a trivial computational task. The following cell implements two simple computational operations. An `inc` function that increments a number by 1, and an `add` function that adds together two numbers. To demonstrate the effects of parallelization, we will also have each one of these functions sleep for a little while, to simulate more intensive time-demanding computations. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "def inc(x):\n",
    "    time.sleep(1)\n",
    "    return x + 1\n",
    "\n",
    "def add(x, y):\n",
    "    time.sleep(1)\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "How much time would it take increment two numbers and add them together?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3 s ± 641 µs per loop (mean ± std. dev. of 7 runs, 1 loop each)\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "x1 = inc(1)\n",
    "x2 = inc(2)\n",
    "z = add(x1, x2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introducing the `delayed` decorator\n",
    "\n",
    "First of all, what is a decorator? One way to think about it (but it's not the [full](https://matthew-brett.github.io/pydagogue/decorating_for_dummies.html) [story](https://matthew-brett.github.io/pydagogue/decorating_for_smart_people.html)) is that it's a function that takes a function as input and produces a function as output. \n",
    "\n",
    "This means that we can process a function that we have written, so that it does something slightly different than it was originally intended to do. \n",
    "\n",
    "In the case of the dask `delayed` decorator, the execution of the function is deferred until a full computational graph can be derived. \n",
    "\n",
    "Let's see what that means in practice. Let's create delayed versions of our `add` and `inc` functions:\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask import delayed\n",
    "\n",
    "delayed_inc = delayed(inc)\n",
    "delayed_add = delayed(add)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "What are these things? "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'dask.delayed.DelayedLeaf'>\n",
      "<class 'dask.delayed.DelayedLeaf'>\n"
     ]
    }
   ],
   "source": [
    "print(type(delayed_inc))\n",
    "print(type(delayed_add))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That's odd. Are these things like a function?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True\n",
      "True\n"
     ]
    }
   ],
   "source": [
    "print(callable(delayed_inc))\n",
    "print(callable(delayed_add))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Looks like they are. What happens when we call them?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "78.4 µs ± 966 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "x1 = delayed_inc(1)\n",
    "x2 = delayed_inc(2)\n",
    "z = delayed_add(x1, x2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Whoa. That was fast! Does `z` take the expected value?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'z' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-7-3a710d2a84f8>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mz\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m: name 'z' is not defined"
     ]
    }
   ],
   "source": [
    "z"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Hmm. Doesn't look like it. In fact, no computation has occurred so far. Dask has only computed a computational graph that it will execute when we call the object's `compute` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "z.visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It's only when we call the `compute` method that the work is done."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%timeit\n",
    "z.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The result is emitted and the computation now takes 2 seconds instead of three! \n",
    "\n",
    "Things to think about: \n",
    "- Why did we go from 3s to 2s? Why weren't we able to parallelize down to 1s?\n",
    "- What would have happened if the inc and add functions didn't include the sleep(1)? Would Dask still be able to speed up this code?\n",
    "- What if we have multiple outputs or also want to get access to x or y?    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise: Parallelize a for loop\n",
    "\n",
    "`for` loops are one of the most common things that we want to parallelize.  Use `dask.delayed` on `inc` and `sum` to parallelize the computation below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = [1, 2, 3, 4, 5, 6, 7, 8]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%timeit\n",
    "# Sequential code\n",
    "\n",
    "results = []\n",
    "for x in data:\n",
    "    y = inc(x)\n",
    "    results.append(y)\n",
    "    \n",
    "total = sum(results)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Your parallel code here..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/01-delayed-loop.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise: Parallelizing a for-loop code with control flow\n",
    "\n",
    "Often we want to delay only *some* functions, running a few of them immediately.  This is especially helpful when those functions are fast and help us to determine what other slower functions we should call.  This decision, to delay or not to delay, is usually where we need to be thoughtful when using `dask.delayed`.\n",
    "\n",
    "In the example below we iterate through a list of inputs.  If that input is even then we want to call `inc`.  If the input is odd then we want to call `double`.  This `is_even` decision to call `inc` or `double` has to be made immediately (not lazily) in order for our graph-building Python code to proceed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def double(x):\n",
    "    sleep(1)\n",
    "    return 2 * x\n",
    "\n",
    "def is_even(x):\n",
    "    return not x % 2\n",
    "\n",
    "data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Sequential code\n",
    "\n",
    "results = []\n",
    "for x in data:\n",
    "    if is_even(x):\n",
    "        y = double(x)\n",
    "    else:\n",
    "        y = inc(x)\n",
    "    results.append(y)\n",
    "    \n",
    "total = sum(results)\n",
    "print(total)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Your parallel code here"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/01-delayed-control-flow.py"
   ]
  }
 ],
 "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}