05_futures.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",
    "# Futures - non-blocking distributed calculations\n",
    "\n",
    "Submit arbitrary functions for computation in a parallelized, eager, and non-blocking way. \n",
    "\n",
    "The `futures` interface (derived from the built-in `concurrent.futures`) provide fine-grained real-time execution for custom situations. We can submit individual functions for evaluation with one set of inputs, or evaluated over a sequence of inputs with `submit()` and `map()`. 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",
    "This is the important difference between futures and delayed. Both can be used to support arbitrary task scheduling, but delayed is lazy (it just constructs a graph) whereas futures are eager. With futures, as soon as the inputs are available and there is compute available, the computation starts. \n",
    "\n",
    "**Related Documentation**\n",
    "\n",
    "* [Futures documentation](https://docs.dask.org/en/latest/futures.html)\n",
    "* [Futures screencast](https://www.youtube.com/watch?v=07EiCpdhtDE)\n",
    "* [Futures examples](https://examples.dask.org/futures.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "\n",
    "client = Client(n_workers=4)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## A Typical Workflow\n",
    "\n",
    "This is the same workflow that we saw in the delayed notebook. It is for-loopy and the data is not necessarily an array or a dataframe. The following example outlines a read-transform-write:\n",
    "\n",
    "```python\n",
    "def process_file(filename):\n",
    "    data = read_a_file(filename)\n",
    "    data = do_a_transformation(data)\n",
    "    destination = f\"results/{filename}\"\n",
    "    write_out_data(data, destination)\n",
    "    return destination\n",
    "\n",
    "futures = []\n",
    "for filename in filenames:\n",
    "    future = client.submit(process_file, filename)\n",
    "    futures.append(future)\n",
    "    \n",
    "futures\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Basics\n",
    "\n",
    "Just like we did in the delayed notebook, 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."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from time import sleep\n",
    "\n",
    "\n",
    "def inc(x):\n",
    "    sleep(1)\n",
    "    return x + 1\n",
    "\n",
    "\n",
    "def double(x):\n",
    "    sleep(2)\n",
    "    return 2 * x\n",
    "\n",
    "\n",
    "def add(x, y):\n",
    "    sleep(1)\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can run these locally"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inc(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Or we can submit them to run remotely with Dask. This immediately returns a future that points to the ongoing computation, and eventually to the stored result."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "future = client.submit(inc, 1)  # returns immediately with pending future\n",
    "future"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you wait a second, and then check on the future again, you’ll see that it has finished."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "future"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can block on the computation and gather the result with the `.result()` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "future.result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Other ways to wait for a future\n",
    "```python\n",
    "from dask.distributed import wait, progress\n",
    "progress(future)\n",
    "```\n",
    "\n",
    "shows a progress bar in *this* notebook, rather than having to go to the dashboard. This progress bar is also asynchronous, and doesn't block the execution of other code in the meanwhile.\n",
    "\n",
    "```python\n",
    "wait(future)\n",
    "```\n",
    "blocks and forces the notebook to wait until the computation pointed to by `future` is done. However, note that if the result of `inc()` is sitting in the cluster, it would take **no time** to execute the computation now, because Dask notices that we are asking for the result of a computation it already knows about. More on this later.\n",
    "\n",
    "#### Other ways to gather results\n",
    "```python\n",
    "client.gather(futures)\n",
    "```\n",
    "\n",
    "gathers results from more than one future."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `client.compute`\n",
    "\n",
    "Generally, any Dask operation that is executed using `.compute()` or `dask.compute()` can be submitted for asynchronous execution using `client.compute()` instead.\n",
    "\n",
    "Here is an example from the delayed notebook:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask\n",
    "\n",
    "\n",
    "@dask.delayed\n",
    "def inc(x):\n",
    "    sleep(1)\n",
    "    return x + 1\n",
    "\n",
    "\n",
    "@dask.delayed\n",
    "def add(x, y):\n",
    "    sleep(1)\n",
    "    return x + y\n",
    "\n",
    "\n",
    "x = inc(1)\n",
    "y = inc(2)\n",
    "z = add(x, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So far we have a regular `dask.delayed` output. When we pass `z` to `client.compute` we get a future back and Dask starts evaluating the task graph. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# notice the difference from z.compute()\n",
    "# notice that this cell completes immediately\n",
    "future = client.compute(z)\n",
    "future"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "future.result()  # waits until result is ready"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When using 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."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `client.submit`\n",
    "\n",
    "`client.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. This looks a lot like doing `client.compute()`, above, except now we are passing the function and arguments directly to the cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def inc(x):\n",
    "    sleep(1)\n",
    "    return x + 1\n",
    "\n",
    "\n",
    "future_x = client.submit(inc, 1)\n",
    "future_y = client.submit(inc, 2)\n",
    "future_z = client.submit(sum, [future_x, future_y])\n",
    "future_z"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "future_z.result()  # waits until result is ready"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The arguments to`client.submit` can be regular Python functions and objects, futures from other submit operations or `dask.delayed` objects."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### How does it work?\n",
    "\n",
    "Each future represents a result held, or being evaluated by the cluster. Thus we can control caching of intermediate values - when a future is no longer referenced, its value is forgotten. In the solution, above, futures are held for each of the function calls. These results would not need to be re-evaluated if we chose to submit more work that needed them.\n",
    "\n",
    "We can explicitly pass data from our local session into the cluster using `client.scatter()`, but usually it is better to construct functions that do the loading of data within the workers themselves, so that there is no need to serialize and communicate the data. Most of the loading functions within Dask, such as `dd.read_csv`, work this way. Similarly, we normally don't want to `gather()` results that are too big in memory."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example: Sporadically failing task\n",
    "\n",
    "Let's imagine a task that sometimes fails. You might encounter this when dealing with input data where sometimes a file is malformed, or maybe a request times out."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from random import random\n",
    "\n",
    "\n",
    "def flaky_inc(i):\n",
    "    if random() < 0.2:\n",
    "        raise ValueError(\"You hit the error!\")\n",
    "    return i + 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you run this function over and over again, it will sometimes fail. \n",
    "\n",
    "```python\n",
    ">>> flaky_inc(2)\n",
    "---------------------------------------------------------------------------\n",
    "ValueError                                Traceback (most recent call last)\n",
    "Input In [65], in <cell line: 1>()\n",
    "----> 1 flaky_inc(2)\n",
    "\n",
    "Input In [61], in flaky_inc(i)\n",
    "      3 def flaky_inc(i):\n",
    "      4     if random() < 0.5:\n",
    "----> 5         raise ValueError(\"You hit the error!\")\n",
    "      6     return i + 1\n",
    "\n",
    "ValueError: You hit the error!\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can run this function on a range of inputs using `client.map`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "futures = client.map(flaky_inc, range(10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice how the cell returned even though some of the computations failed. We can inspect these futures one by one and find the ones that failed:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for i, future in enumerate(futures):\n",
    "    print(i, future.status)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can rerun those specific futures to try to get the task to successfully complete:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "futures[5].retry()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for i, future in enumerate(futures):\n",
    "    print(i, future.status)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A more concise way of retrying in the case of sporadic failures is by setting the number of retries in the `client.compute`, `client.submit` or `client.map` method.\n",
    "\n",
    "**Note**: In this example we also need to set `pure=False` to let Dask know that the arguments to the function do not totally determine the output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "futures = client.map(flaky_inc, range(10), retries=5, pure=False)\n",
    "future_z = client.submit(sum, futures)\n",
    "future_z.result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You will see a lot of warnings, but the computation should eventually succeed."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Why use Futures?\n",
    "\n",
    "The futures API offers a work submission style that can easily emulate the map/reduce paradigm. If that is familiar to you then futures might be the simplest entrypoint into Dask. \n",
    "\n",
    "The other big benefit of futures is that the intermediate results, represented by futures, can be passed to new tasks without having to pull data locally from the cluster. New operations can be setup to work on the output of previous jobs that haven't even begun yet."
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}