04_distributed.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"images/dask_horizontal.svg\" align=\"right\" width=\"30%\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Table of Contents\n",
    "* [Distributed](#Distributed)\n",
    "\t* [Making a cluster](#Making-a-cluster)\n",
    "\t\t* [Detailed method](#Detailed-method)\n",
    "\t\t* [Simple method](#Simple-method)\n",
    "\t* [Executing with the distributed client](#Executing-with-the-distributed-client)\n",
    "\t* [Distributed futures](#Distributed-futures)\n",
    "\t* [Debugging](#Debugging)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Distributed"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we saw in Foundations, dask allows you to simply construct graphs of tasks with dependencies. In fact, if you skip forward, you will find that graphs can also be created automatically for you using functional, Numpy or Pandas syntax on data collections. None of this would be very useful, if there weren't also a way to execute these graphs, in a parallel and memory-aware way. Dask comes with four available schedulers:\n",
    "- dask.threaded.get: a scheduler backed by a thread pool\n",
    "- dask.multiprocessing.get: a scheduler backed by a process pool\n",
    "- dask.async.get_sync: a synchronous scheduler, good for debugging\n",
    "- distributed.Client.get: a distributed scheduler for executing graphs on multiple machines.\n",
    "\n",
    "To select one of these for computation, you can specify at the time of asking for a result\n",
    "```python\n",
    "myvalue.compute(get=dask.async.get_sync)  # for debugging\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "or set the current default, either temporarily or globally\n",
    "```python\n",
    "with dask.set_options(get=dask.multiprocessing.get):\n",
    "    # set temporarily fo this block only\n",
    "    myvalue.compute()\n",
    "\n",
    "dask.set_options(get=dask.multiprocessing.get)\n",
    "# set until further notice\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For single-machine use, the threaded and multiprocessing schedulers are fine choices. However, for scaling out work across a cluster, the distributed scheduler is required. Indeed, this is now generally preferred for all work, because it gives you additional monitoring information not available in the other schedulers. (Some of this monitoring is also available with an explicit progress bar and profiler, see [here](http://dask.pydata.org/en/latest/diagnostics.html).)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Making a cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Detailed method"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following process explains what happens under the hood when setting up a computation environment with the dask distributed scheduler *by hand*. It is not necessary to do this for the rest of the tutorial, but understanding what is going on will help a great deal when scaling up computations across a cluster. Users may wish to skip this section for now, and continue with the Simple method, below.\n",
    "\n",
    "**The scheduler**\n",
    "\n",
    "In a terminal, type the following:\n",
    "```\n",
    "dask-scheduler\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You will get text something like the following:\n",
    "```\n",
    "distributed.scheduler - INFO - -----------------------------------------------\n",
    "distributed.scheduler - INFO -   Scheduler at:         192.168.0.11:8786\n",
    "distributed.scheduler - INFO -       bokeh at:         192.168.0.11:8788\n",
    "distributed.scheduler - INFO -        http at:         192.168.0.11:9786\n",
    "distributed.bokeh.application - INFO - Web UI: http://192.168.0.11:8787/status/\n",
    "distributed.scheduler - INFO - -----------------------------------------------\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The top line gives the address, 192.168.0.1:8786, at which the scheduler is waiting for connections - it is this address that workers and clients need to be given (your IP and/or port numbers may be different). The further addresses are for an in-process bokeh graph server for scheduler debugging, a JSON http endpoint for information about the server, and, finally, the URL of the main monitoring dashboard; you can type this into a web-browser, but it will not show much information yet.\n",
    "\n",
    "The scheduler cannot do much without workers. We can create a worker process with:\n",
    "```\n",
    "dask-worker 192.168.0.11:8786\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "where the address should be the same as given by the scheduler process, above. By default, the worker will start a monitoring process (the *nanny*), and a worker process with the number of threads equal to the number of cores (all the values can be changed). The worker has its own http and bokeh server. From the text displayed in the console, we see that the worker connects to the scheduler - information is also printed by the scheduler indicating that it has received a connection from a worker. Notice that this worker process could have been on a different machine from the scheduler.\n",
    "\n",
    "Next, in a new python session (perhaps in the notebook, or another console, we can do\n",
    "```python\n",
    "from dask.distributed import Client\n",
    "c = Client('192.168.0.11:8786')\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "to connect to the scheduler. Again, the address must match the scheduler, above, and that, again, the scheduler logs the connection from the client. This client is now ready to accept work, and coordinate with the scheduler such that tasks get executed by the threads of the worker process."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The three python-running consoles might look something like the following:\n",
    "![distributed session](images/distributed_session.png)\n",
    "\n",
    "Note that both the scheduler and worker commands accept a number of parameters to define the ports used, the number of threads/processed, memory limits, etc. - these will become useful when customising deployments.\n",
    "\n",
    "A similar method can be used to set up the scheduler and workers across a number of cluster nodes, and connect to them from a client to do work. There are some automated options for achieving this, including for resource management and dynamic clustering scenarios, see [here](http://distributed.readthedocs.io/en/latest/setup.html)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Simple method"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Throughout the rest of this tutorial, we will be using the default dask distributed cluster. This gets created automatically when creating a client with no arguments, if no client has yet been defined. Creating any distributed client also sets it to be the default executor of dask `compute` calls, unless otherwise specified."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# be sure to shut down other kernels running distributed clients\n",
    "from dask.distributed import Client\n",
    "c = Client()\n",
    "c"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The scheduler is now listening on your laptop, and has a number of worker processes connected. Furthermore, the web UI will be available on `127.0.0.1:8787/status` - you can open this in a new tab of your browser. Other monitoring output is also available, e.g., `/tasks`.\n",
    "\n",
    "Note that you should close any other open kernels using a distributed cluster created this way, because otherwise the new one will not be able to use port 8787, and you will not be able to access the monitoring dashboard.\n",
    "\n",
    "![ui](images/ui.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "No tasks are yet being processed, and no data is held in the memory of the workers, so the lower part of the display is empty for the moment."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Executing with the distributed client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Consider some trivial calculation, as in previous sections, where we have added sleep statements in order to simulate real work being done."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from dask import delayed\n",
    "import time\n",
    "\n",
    "def inc(x):\n",
    "    time.sleep(5)\n",
    "    return x + 1\n",
    "\n",
    "def dec(x):\n",
    "    time.sleep(3)\n",
    "    return x - 1\n",
    "\n",
    "def add(x, y):\n",
    "    time.sleep(7)\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "x = delayed(inc)(1)\n",
    "y = delayed(dec)(2)\n",
    "total = delayed(add)(x, y)\n",
    "total.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The tasks will appear in the web UI as they are processed by the cluster and, eventually, a result will be printed as output of the cell above. Note that the kernel is blocked while waiting for the result. The resulting tasks block graph might look something like below. Hovering over each block gives which function it related to, and how long it took to execute. ![this](images/tasks.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If all you want to do is execute computations created using delayed, or run calculations based on the higher-level data collections (see the coming sections), then that is about all you need to know to scale your work up to cluster scale. However, there is more detail to know about the distributed scheduler that will help with efficient usage. See the chapter Distributed, Advanced."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [default]",
   "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.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}