06-distributed-advanced.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",
    "# The async futures interface\n",
    "\n",
    "In previous notebooks we showed that the distributed scheduler can be used to execute dask graphs, and so compute collections across multiple machines. However, the distributed scheduler also offers additional API that allow for finer control and asynchronous computation. In this notebook we delve into these features.\n",
    "\n",
    "To begin, the `distributed` scheduler implements a superset of of the standard library [`concurrent.futures`](https://docs.python.org/3/library/concurrent.futures.html) API, allowing for asynchronous map-reduce like functionality. We can submit individual functions for evaluation with one set of inputs, or evaluated over a sequence of inputs with `submit()` and `map()`. Notice that the call returns immediately, giving one or more *futures*, whose status begins as \"pending\" and later becomes \"finished\". There is no blocking of the local python session.\n",
    "\n",
    "Before you start, you should open the diagnostics dashboard so you can watch what the scheduler is doing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from scipy_utils import make_cluster\n",
    "from dask.distributed import Client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster = make_cluster()\n",
    "cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c = Client(cluster)\n",
    "c"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define some toy functions for simulating work\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": "markdown",
   "metadata": {},
   "source": [
    "### `Client.submit`\n",
    "\n",
    "`submit` takes a function and arguments, pushes these to the cluster, returning a *Future* representing the result to be computed. The function is passed to a worker process for evaluation. Note that this cell returns immediately, while computation may still be ongoing on the cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fut = c.submit(inc, 1)\n",
    "fut"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Looking at the `repr` of the future, `fut`, you can see it's status is \"pending\". This means that the computation the future represents hasn't finished yet. Since this is asynchronous, you can do other things while it computes. However, if you want to wait until the future has completed you can use the `wait` function. At this time the future's status will become \"finished\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from distributed import wait"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Block until `fut` has completed\n",
    "wait(fut);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To retrieve a result you can use the `.result` method (which fetches for the one future), or `Client.gather` (which fetches for one or more futures). In both cases these methods will block until all requested futures have completed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Retrieve the data using `.result()`\n",
    "fut.result()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Retrieve the data using `Client.gather` - result is now already available\n",
    "c.gather(fut)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here we see an alternative way to execute work on the cluster: when you submit or map with the inputs as futures, the *computation moves to the data* rather than the other way around, and the client, in the local python session, need never see the intermediate values. This is similar to building the graph using delayed, and indeed, delayed can be used in conjunction with futures. Here we use the delayed object `total` from before."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask import delayed\n",
    "\n",
    "x = delayed(inc)(1)\n",
    "y = delayed(dec)(2)\n",
    "total = delayed(add)(x, y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# notice the difference from total.compute()\n",
    "# notice that this cell completes immediately\n",
    "fut = c.compute(total)\n",
    "fut"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.gather(fut)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise: Rebuild the above delayed computation using `Client.submit` instead"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Solution:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/client_submit.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The futures API offers a work submission style that can easily emulate the map/reduce paradigm (see `c.map()`) that may be familiar. The intermediate results, represented by futures, can be passed to new tasks without having to bring the data locally from the cluster, and new work can be assigned using the output of previous jobs that haven't begun yet.\n",
    "\n",
    "Generally, any dask operation that is executed using `.compute()` can be submitted for asynchronous execution using `c.compute()` instead, and this applies to all collections. Here we create a `Bag`, do some operations on it, and call `Client.compute` on it. Since this is asynchronous we could continue to submit more work (perhaps based on the result of the calculation), or, as in the next cell, follow the progress of the computation. A similar progress-bar appears in the monitoring UI page."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.bag as db\n",
    "\n",
    "res = (db.from_sequence(range(10))\n",
    "         .map(inc)\n",
    "         .filter(lambda x: x % 2 == 0)\n",
    "         .sum())\n",
    "\n",
    "res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "f = c.compute(res)\n",
    "f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from distributed.diagnostics import progress"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# note that progress must be the last line of a cell\n",
    "# in order to show up\n",
    "progress(f)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.gather(f)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Note well**: a future points to data being computed on the cluster or held in memory. To release the memory, all references to a future need to be deleted. Look at the scheduler dashboard to see the effect of the following."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "del f, fut"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Asynchronous computation\n",
    "<img style=\"float: right;\" src=\"https://upload.wikimedia.org/wikipedia/commons/thumb/3/32/Rosenbrock_function.svg/450px-Rosenbrock_function.svg.png\" height=200 width=200>\n",
    "\n",
    "One benefit of using the futures API is that you can have dynamic computations that adjust as things progress. Here we implement a simple naive search by looping through results as they come in, and submit new points to compute as others are still running.\n",
    "\n",
    "Watching the [diagnostics dashboard](../../9002/status) as this runs you can see computations are being concurrently run while more are being submitted. This flexibility can be useful for parallel algorithms that require some level of synchronization."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# a simple function with interesting minima\n",
    "\n",
    "def rosenbrock(point):\n",
    "    \"\"\"Compute the rosenbrock function and return the point and result\"\"\"\n",
    "    time.sleep(0.1)\n",
    "    score = (1 - point[0])**2 + 2 * (point[1] - point[0]**2)**2\n",
    "    return point, score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from bokeh.io import output_notebook, push_notebook\n",
    "from bokeh.models.sources import ColumnDataSource\n",
    "from bokeh.plotting import figure, show\n",
    "import numpy as np\n",
    "output_notebook()\n",
    "\n",
    "# set up plot background\n",
    "N = 500\n",
    "x = np.linspace(-5, 5, N)\n",
    "y = np.linspace(-5, 5, N)\n",
    "xx, yy = np.meshgrid(x, y)\n",
    "d = (1 - xx)**2 + 2 * (yy - xx**2)**2\n",
    "d = np.log(d)\n",
    "\n",
    "p = figure(x_range=(-5, 5), y_range=(-5, 5))\n",
    "p.image(image=[d], x=-5, y=-5, dw=10, dh=10, palette=\"Spectral11\");"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import as_completed\n",
    "from random import uniform\n",
    "\n",
    "scale = 5                  # Intial random perturbation scale\n",
    "best_point = (0, 0)        # Initial guess\n",
    "best_score = float('inf')  # Best score so far\n",
    "startx = [uniform(-scale, scale) for _ in range(10)]\n",
    "starty = [uniform(-scale, scale) for _ in range(10)]\n",
    "\n",
    "# set up plot\n",
    "source = ColumnDataSource({'x': startx, 'y': starty, 'c': ['grey'] * 10})\n",
    "p.circle(source=source, x='x', y='y', color='c')\n",
    "t = show(p, notebook_handle=True)\n",
    "\n",
    "# initial 10 random points\n",
    "futures = [c.submit(rosenbrock, (x, y)) for x, y in zip(startx, starty)]\n",
    "iterator = as_completed(futures)\n",
    "\n",
    "for res in iterator:\n",
    "    # take a completed point, is it an improvement?\n",
    "    point, score = res.result()\n",
    "    if score < best_score:\n",
    "        best_score, best_point = score, point\n",
    "        print(score, point)\n",
    "\n",
    "    x, y = best_point\n",
    "    newx, newy = (x + uniform(-scale, scale), y + uniform(-scale, scale))\n",
    "    \n",
    "    # update plot\n",
    "    source.stream({'x': [newx], 'y': [newy], 'c': ['grey']}, rollover=20)\n",
    "    push_notebook(t)\n",
    "    \n",
    "    # add new point, dynamically, to work on the cluster\n",
    "    new_point = c.submit(rosenbrock, (newx, newy))\n",
    "    iterator.add(new_point)  # Start tracking new task as well\n",
    "\n",
    "    # Narrow search and consider stopping\n",
    "    scale *= 0.99\n",
    "    if scale < 0.001:\n",
    "        break\n",
    "point"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# clean up\n",
    "del futures[:], new_point, iterator, res\n",
    "client.close()\n",
    "cluster.close()"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}