02-delayed.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"http://dask.readthedocs.io/en/latest/_images/dask_horizontal.svg\"\n",
    "     align=\"right\"\n",
    "     width=\"30%\"\n",
    "     alt=\"Dask logo\\\">\n",
    "\n",
    "# Parallelize code with `dask.delayed`\n",
    "\n",
    "In this section we parallelize for-loop style code with Dask Delayed.  This approach is more flexible and more manual than automatic approaches like Dask Dataframe.  It is commonly useful to parallelize existing codebases, or to build tools like Dask Dataframe.\n",
    "\n",
    "To see a real-world application you may with to read about [Credit Modeling with Dask](https://blog.dask.org/2018/02/09/credit-models-with-dask).  This will also help us to develop an understanding for Dask in general.\n",
    "\n",
    "In this notebook we start with toy examples to build understanding.  Then we end with two examples that build a dataframe and machine learning algorithm.\n",
    "\n",
    "Objectives:\n",
    "\n",
    "1.  Use Dask delayed to build custom task graphs\n",
    "2.  Build an intuition for computational task scheduling generally\n",
    "3.  Learn how tools like Dask Dataframe and Dask-ML work internally"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Basics\n",
    "\n",
    "First let's make some toy functions, `inc` and `add`, that sleep for a while to simulate work. We'll then time running these functions normally.\n",
    "\n",
    "In the next section we'll parallelize this code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from time import sleep\n",
    "\n",
    "def inc(x):\n",
    "    sleep(1)\n",
    "    return x + 1\n",
    "\n",
    "def add(x, y):\n",
    "    sleep(1)\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We time the execution of this normal code using the `%%time` magic, which is a special function of the Jupyter Notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# This takes three seconds to run because we call each\n",
    "# function sequentially, one after the other\n",
    "\n",
    "x = inc(1)\n",
    "y = inc(2)\n",
    "z = add(x, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Parallelize with the `dask.delayed` decorator\n",
    "\n",
    "Those two increment calls *could* be called in parallel because they are independent of one-another.\n",
    "\n",
    "We'll transform the `inc` and `add` functions using the `dask.delayed` function. When we call the delayed version by passing the arguments, exactly as before, but the original function isn't actually called yet - which is why the cell execution finishes very quickly.\n",
    "Instead, a *delayed* object is made, which keeps track of the function to call and the arguments to pass to it.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "# This executes actual code sequentially\n",
    "\n",
    "# x = inc(1)\n",
    "# y = inc(2)\n",
    "# z = add(x, y)\n",
    "\n",
    "# This runs lazily, all it does is build a graph\n",
    "\n",
    "x = dask.delayed(inc)(1)\n",
    "y = dask.delayed(inc)(2)\n",
    "z = dask.delayed(add)(x, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This ran immediately, since nothing has really happened yet.\n",
    "\n",
    "To get the result, call `compute`. Notice that this runs faster than the original code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "# This actually runs our computation using a local thread pool\n",
    "z.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## What just happened?\n",
    "\n",
    "The `z` object is a lazy `Delayed` object.  This object holds everything we need to compute the final result, including references to all of the functions that are required and their inputs and relationship to one-another.  We can evaluate the result with `.compute()` as above or we can visualize the task graph for this value with `.visualize()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "z"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Look at the task graph for `z`\n",
    "z.visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice that this includes the names of the functions from before, and the logical flow of the outputs of the `inc` functions to the inputs of `add`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Some questions to consider:\n",
    "\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": [
    "%%time\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": [
    "How do the graph visualizations compare with the given solution, compared to a version with the `sum` function used directly rather than wrapped with `delay`? Can you explain the latter version? You might find the result of the following expression illuminating\n",
    "```python\n",
    "delayed(inc)(1) + delayed(inc)(2)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Start a Dask Cluster on Kubernetes\n",
    "\n",
    "The code above ran in threads in your local Jupyter notebook.  Now we'll now start up a Dask cluster on Kubernetes.\n",
    "\n",
    "*Note: if you still have a cluster running in your other notebook you may want to close that cluster, or restart your notebook.  Otherwise your dashboard will point to the previous cluster.*\n",
    "\n",
    "    cluster.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_kubernetes import KubeCluster\n",
    "cluster = KubeCluster(n_workers=20)\n",
    "cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "client = Client(cluster)"
   ]
  },
  {
   "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 inc(x):\n",
    "    sleep(1)\n",
    "    return x + 1\n",
    "\n",
    "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...\n",
    "# TODO: parallelize the sequential code above using dask.delayed\n",
    "# You will need to delay some functions, but not all"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/01-delayed-control-flow.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time total.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total.visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Some questions to consider:\n",
    "\n",
    "-  What are other examples of control flow where we can't use delayed?\n",
    "-  What would have happened if we had delayed the evaluation of `is_even(x)` in the example above?\n",
    "-  What are your thoughts on delaying `sum`?  This function is both computational but also fast to run."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Rebuild Dataframe algorithms manually\n",
    "\n",
    "In the last notebook we used Dask Dataframe to load CSV data from the cloud and then perform some basic analyses.  In these examples Dask dataframe automatically built our parallel algorithms for us.\n",
    "\n",
    "In this section we'll do that same work, but now we'll use Dask delayed to construct these algorithms manually.  In practice you don't have to do this because Dask dataframe already exists, but doing it once, manually can help you understand both how Dask dataframe works, and how to parallelize your own code.\n",
    "\n",
    "To make things a bit faster we've also decided to store data in the Parquet format.  We'll use Dask delayed along with Arrow to read this data in many small parts, convert those parts to Pandas dataframes, and then do a groupby-aggregation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inspect Parquet data on the cloud\n",
    "\n",
    "Just like our CSV files, we also have many parquet files for the same data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import gcsfs\n",
    "gcs = gcsfs.GCSFileSystem()\n",
    "filenames = sorted(gcs.glob('anaconda-public-data/nyc-taxi/nyc.parquet/part.*.parquet'))\n",
    "filenames[:5]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can read that data using the [gcsfs](https://gcsfs.readthedocs.io/en/latest/) library to access data on Google Cloud Storage, and either [PyArrow](https://arrow.apache.org/docs/python/) or [Fastparquet](https://fastparquet.readthedocs.io/en/latest/) to read those bytes.  \n",
    "\n",
    "Here we'll use Arrow to turn one Parquet file into one Pandas dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyarrow.parquet as pq\n",
    "\n",
    "fn = filenames[0]\n",
    "with gcs.open(fn) as f:\n",
    "    pf = pq.ParquetFile(f)  # Arrow ParquetFile\n",
    "    table = pf.read()       # Arrow Table\n",
    "    df = table.to_pandas()  # Pandas DataFrame\n",
    "\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Build aggregation piece by piece\n",
    "\n",
    "We want to compute the following operation on all of our data:\n",
    "\n",
    "```python\n",
    "import dask.dataframe as dd\n",
    "df = dd.read_parquet('gcs://anaconda-public-data/nyc-taxi/nyc.parquet/part.*.parquet')\n",
    "df.passenger_count.mean().compute()\n",
    "```\n",
    "\n",
    "This actually works, but lets pretend that it didn't, and lets build this up, chunk by chunk, file by file.\n",
    "\n",
    "We do this for you sequentially below with a for loop:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sums = []\n",
    "counts = []\n",
    "\n",
    "def filename_to_dataframe(fn):\n",
    "    gcs = gcsfs.GCSFileSystem()\n",
    "\n",
    "    with gcs.open(fn) as f:\n",
    "        pf = pq.ParquetFile(f)  # Arrow ParquetFile\n",
    "        table = pf.read()       # Arrow Table\n",
    "        df = table.to_pandas()  # Pandas DataFrame\n",
    "\n",
    "    return df\n",
    "\n",
    "\n",
    "for fn in filenames[:3]:\n",
    "\n",
    "    # Read in parquet file to Pandas Dataframe\n",
    "    df = filename_to_dataframe(fn)\n",
    "    \n",
    "    # Groupby origin airport\n",
    "    total = df.passenger_count.sum()\n",
    "    \n",
    "    # Number of flights by origin\n",
    "    count = df.passenger_count.count()\n",
    "    \n",
    "    # Save the intermediates\n",
    "    sums.append(total)\n",
    "    counts.append(count)\n",
    "\n",
    "# Combine intermediates to get total mean-delay-per-origin\n",
    "total_sums = sum(sums)\n",
    "total_counts = sum(counts)\n",
    "mean = total_sums / total_counts\n",
    "mean"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Parallelize the code above\n",
    "\n",
    "Use `dask.delayed` to parallelize the code above.  Some extra things you will need to know.\n",
    "\n",
    "1.  Methods and attribute access on delayed objects work automatically, so if you have a delayed object you can perform normal arithmetic, slicing, and method calls on it and it will produce the correct delayed calls.\n",
    "\n",
    "    ```python\n",
    "    x = dask.delayed(np.arange)(10)\n",
    "    y = (x + 1)[::2].sum()  # everything here was delayed\n",
    "    ```\n",
    "\n",
    "So your goal is to parallelize the code above (which has been copied below) using `dask.delayed`.  You may also want to visualize a bit of the computation to see if you're doing it correctly."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "# copied sequential code\n",
    "sums = []\n",
    "counts = []\n",
    "\n",
    "def filename_to_dataframe(fn):\n",
    "    gcs = gcsfs.GCSFileSystem()\n",
    "\n",
    "    with gcs.open(fn) as f:\n",
    "        pf = pq.ParquetFile(f)  # Arrow ParquetFile\n",
    "        table = pf.read()       # Arrow Table\n",
    "        df = table.to_pandas()  # Pandas DataFrame\n",
    "\n",
    "    return df\n",
    "\n",
    "\n",
    "for fn in filenames[:3]:\n",
    "\n",
    "    # Read in parquet file to Pandas Dataframe\n",
    "    df = filename_to_dataframe(fn)\n",
    "    \n",
    "    # Groupby origin airport\n",
    "    total = df.passenger_count.sum()\n",
    "    \n",
    "    # Number of flights by origin\n",
    "    count = df.passenger_count.count()\n",
    "    \n",
    "    # Save the intermediates\n",
    "    sums.append(total)\n",
    "    counts.append(count)\n",
    "\n",
    "# Combine intermediates to get total mean-delay-per-origin\n",
    "total_sums = sum(sums)\n",
    "total_counts = sum(counts)\n",
    "mean = total_sums / total_counts\n",
    "mean"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you load the solution, add `%%time` to the top of the cell to measure the running time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/01-delayed-dataframe.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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}