04-schedulers.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",
    "\n",
    "## Schedulers\n",
    "\n",
    "In the previous notebooks, we used `dask.delayed` and `dask.dataframe` to parallelize computations.\n",
    "These work by building a *task graph* instead of executing immediately.\n",
    "Each *task* represents some function to call on some data, and the full *graph* is the relationship between all the tasks.\n",
    "\n",
    "When we wanted the actual result, we called `compute`, which handed the task graph off to a *scheduler*.\n",
    "\n",
    "**Schedulers are responsible for running a task graph and producing a result**.\n",
    "\n",
    "![](https://raw.githubusercontent.com/dask/dask-org/master/images/grid_search_schedule.gif)\n",
    "\n",
    "First, there are the single machine schedulers that execute things in parallel using threads or processes (or synchronously for debugging). These are what we've used up until now. Second, there's the `dask.distributed` scheduler, which is newer and has more features than the single machine scheduler.\n",
    "\n",
    "In this notebook we'll first talk about the different schedulers. Then we'll use the `dask.distributed` scheduler in more depth."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Local Schedulers\n",
    "\n",
    "Dask separates computation description (task graphs) from execution (schedulers). This allows you to write code once, and run it locally or scale it out across a cluster.\n",
    "\n",
    "Here we discuss the *local* schedulers - schedulers that run only on a single machine. The three options here are:\n",
    "\n",
    "- `dask.threaded.get         # uses a local thread pool`\n",
    "- `dask.multiprocessing.get  # uses a local process pool`\n",
    "- `dask.get                  # uses only the main thread (useful for debugging)`\n",
    "\n",
    "In each case we change the scheduler used in a few different ways:\n",
    "\n",
    "- By providing a `get=` keyword argument to `compute`:\n",
    "\n",
    "```python\n",
    "total.compute(get=dask.multiprocessing.get)\n",
    "# or \n",
    "dask.compute(a, b, get=dask.multiprocessing.get)\n",
    "```\n",
    "\n",
    "- Using `dask.set_options`:\n",
    "\n",
    "```python\n",
    "# Use multiprocessing in this block\n",
    "with dask.set_options(get=dask.multiprocessing.get):\n",
    "    total.compute()\n",
    "# Use multiprocessing globally\n",
    "dask.set_options(get=dask.multiprocessing.get)\n",
    "```\n",
    "\n",
    "---\n",
    "\n",
    "*Note: on master, we've also added a `scheduler=` keyword to `compute` that takes in scheduler names instead of scheduler functions. In future releases `dask.compute(..., scheduler='threads')` or `dask.set_options(scheduler='threads')` will be the preferred methods.*\n",
    "\n",
    "---\n",
    "\n",
    "Here we repeat a simple dataframe computation from the previous section using the different schedulers:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import dask.dataframe as dd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = dd.read_csv(os.path.join('data', 'nycflights', '*.csv'),\n",
    "                 parse_dates={'Date': [0, 1, 2]},\n",
    "                 dtype={'TailNum': object,\n",
    "                        'CRSElapsedTime': float,\n",
    "                        'Cancelled': bool})\n",
    "\n",
    "# Maximum non-cancelled delay\n",
    "largest_delay = df[~df.Cancelled].DepDelay.max()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "largest_delay"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time _ = largest_delay.compute()  # this uses threads by default"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.multiprocessing\n",
    "%time _ = largest_delay.compute(get=dask.multiprocessing.get)  # this uses processes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time _ = largest_delay.compute(get=dask.get)  # This uses a single thread"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "By default the threaded and multiprocessing schedulers use the same number of workers as cores. You can change this using the `num_workers` keyword in the same way that you specified `get` above:\n",
    "\n",
    "```\n",
    "largest_delay.compute(get=dask.multiprocessing.get, num_workers=2)\n",
    "```\n",
    "\n",
    "To see how many cores you have on your computer, you can use `multiprocessing.cpu_count`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from multiprocessing import cpu_count\n",
    "cpu_count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Some Questions to Consider:\n",
    "\n",
    "- How much speedup is possible for this task (hint, look at the graph).\n",
    "- Given how many cores are on this machine, how much faster could the parallel schedulers be than the single-threaded scheduler.\n",
    "- How much faster was using threads over a single thread? Why does this differ from the optimal speedup?\n",
    "- Why is the multiprocessing scheduler so much slower here?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## In what cases would you want to use one scheduler over another?\n",
    "\n",
    "http://dask.pydata.org/en/latest/setup/single-machine.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Distributed Scheduler\n",
    "\n",
    "The `dask.distributed` system is composed of a single centralized scheduler and many worker processes. [Deploying](http://dask.pydata.org/en/latest/setup.html) a remote Dask cluster involves some additional effort. But doing things locally is just involves creating a `Client` object, which lets you interact with the \"cluster\" (local threads or processes on your machine). For more information see [here](http://dask.pydata.org/en/latest/setup/single-distributed.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "\n",
    "# Setup a local cluster.\n",
    "# By default this sets up 1 worker per core\n",
    "client = Client()\n",
    "client.cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Be sure to click the `Dashboard` link to open up the diagnostics dashboard.\n",
    "\n",
    "By default, creating a `Client` makes it the default scheduler. Any calls to `.compute` will use the cluster your `client` is attached to (See http://dask.pydata.org/en/latest/scheduling.html for how to specify which scheduler to use)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time largest_delay.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Some Questions to Consider\n",
    "\n",
    "- How does this compare to the optimal parallel speedup?\n",
    "- Why is this faster than the threaded scheduler?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise\n",
    "\n",
    "Run the following computations while looking at the diagnostics page. In each case what is taking the most time?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Number of flights\n",
    "_ = len(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Number of non-cancelled flights\n",
    "_ = len(df[~df.Cancelled])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Number of non-cancelled flights per-airport\n",
    "_ = df[~df.Cancelled].groupby('Origin').Origin.count().compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Average departure delay from each airport?\n",
    "_ = df[~df.Cancelled].groupby('Origin').DepDelay.mean().compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Average departure delay per day-of-week\n",
    "_ = df.groupby(df.Date.dt.dayofweek).DepDelay.mean().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### New API\n",
    "\n",
    "The distributed scheduler is more sophisticated than the single machine schedulers. It can compute asynchronously, and also provides an api similar to that of `concurrent.futures`. This will be discussed more [in part 6](06-distributed-advanced.ipynb). For further information you can also see the docs http://distributed.readthedocs.io/en/latest/."
   ]
  }
 ],
 "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.5.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}