07-Bag-Future.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Bags, Futures, and Some Bonus Features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this module, we'll take a *really* quick look at some additional aspects of Dask. The goal is not to make you proficient, or even talk about the full scope of what's possible, but to give you some orientation and awareness.\n",
    "\n",
    "That way, if you're working on a project where these features might come in handy, you'll at least know where to look (or what to Google) to learn more."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "\n",
    "client = Client(n_workers=2, threads_per_worker=1, memory_limit='256MB')\n",
    "\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dask Bag\n",
    "\n",
    "Dask's __Bag__ interface is the last of the high-level interfaces for manipulating data with Dask.\n",
    "\n",
    "A __Bag__ is an unordered collection that *can* contain the same item multiple times (unlike a Python Set, which cannot distinguish duplicates). A __Bag__ is a little bit like a Python Counter. \n",
    "\n",
    "But its main feature is that it can be partitioned and processed in parallel.\n",
    "\n",
    "Bags are documented at https://docs.dask.org/en/latest/bag.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.bag as db\n",
    "\n",
    "b = db.from_sequence(range(30), npartitions=4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "b"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "b.count().compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "b.npartitions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We program bags using functional-programming abstractions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "b_squared = b.map(lambda x:x**2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "b_squared.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "even_odd_groups = b_squared.groupby(lambda x: x%2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "even_odd_groups.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "even_odd_groups.take(1)[0][1]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Remember, bags aren't ordered -- so you may or may not see the output in ascending sequence. \n",
    "\n",
    "Also, as with many data-parallel collection frameworks, we need need to be careful about \"materializing groups\" or \"touching the data\" ... In this case, we've asked Dask to fully compute the two groups (odd and even squares). \n",
    "\n",
    "It's a silly example, but we definitely want to be careful about this, since on real large datasets, it's common for individual groups to be too large to fit in memory (imagine taking your company's transactions and grouping by country!)\n",
    "\n",
    "Whenever you think \"group-by\" (and maybe if you're thinking group-by you should think about dataframe instead) it's good to think, \"... and then what?\" ...\n",
    "\n",
    "Typically, group-by goes with an aggregation like \"group-by and count\" or \"group-by and add\" -- in those cases, use the alternative operation `foldby`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def add(a,b):\n",
    "    return a+b\n",
    "\n",
    "b_squared.foldby(lambda n:n%2, add, 0, add, 0).compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def count(total, item):\n",
    "    return total+1\n",
    "\n",
    "b_squared.foldby(lambda n:n%2, count, 0, add, 0).compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Again, this is just to demonstrate the similarity between bag and other unstructured data representations in data-parallel tools, not to encourage their use when other approaches make more sense."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise: Dask Delayed and `concurrent.futures`\n",
    "\n",
    "*Work through these excellent examples courtesy of Dask contributor James Bourbeau*\n",
    "\n",
    "*More detailed docs are online at:*\n",
    "* https://docs.dask.org/en/latest/delayed.html\n",
    "* https://docs.dask.org/en/latest/futures.html\n",
    "\n",
    "Sometimes problems don’t fit nicely into one of the high-level collections like Dask arrays or Dask DataFrames. In these cases, you can parallelize custom algorithms using the Dask `delayed` interface. This allows one to manually create task graphs with a light annotation of normal Python code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "def inc(x):\n",
    "    time.sleep(0.5)\n",
    "    return x + 1\n",
    "\n",
    "def double(x):\n",
    "    time.sleep(0.5)\n",
    "    return 2 * x\n",
    "\n",
    "def add(x, y):\n",
    "    time.sleep(0.5)\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "data = [1, 2, 3, 4]\n",
    "\n",
    "output = []\n",
    "for x in data:\n",
    "    a = inc(x)\n",
    "    b = double(x)\n",
    "    c = add(a, b)\n",
    "    output.append(c)\n",
    "\n",
    "total = sum(output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask `delayed` wraps function calls and delays their execution. Rather than computing results immediately, `delayed` functions record what we want to compute as a task into a graph that we’ll run later on parallel hardware by calling `compute`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask import delayed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lazy_inc = delayed(inc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inc_output = lazy_inc(3)  #inc(3)\n",
    "inc_output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inc_output.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inc_output.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using delayed functions, we can build up a task graph for the particular computation we want to perform"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "double_inc_output = lazy_inc(inc_output)\n",
    "double_inc_output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "double_inc_output.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "double_inc_output.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can use `delayed` to make our previous example computation lazy by wrapping all the function calls with delayed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@delayed\n",
    "def inc(x):\n",
    "    time.sleep(0.5)\n",
    "    return x + 1\n",
    "\n",
    "@delayed\n",
    "def double(x):\n",
    "    time.sleep(0.5)\n",
    "    return 2 * x\n",
    "\n",
    "@delayed\n",
    "def add(x, y):\n",
    "    time.sleep(0.5)\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now `add` returns a `Delayed` object which you can call `compute()` on at a later time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "data = [1, 2, 3, 4]\n",
    "\n",
    "output = []\n",
    "for x in data:\n",
    "    a = inc(x)\n",
    "    b = double(x)\n",
    "    c = add(a, b)\n",
    "    output.append(c)\n",
    "\n",
    "total = delayed(sum)(output)\n",
    "total"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "total.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check out the [Dask delayed best practices](http://docs.dask.org/en/latest/delayed-best-practices.html) page to avoid some common problems when using `delayed`. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `concurrent.futures` interface\n",
    "\n",
    "The Dask distributed scheduler implements a superset of Python's [`concurrent.futures`](https://docs.python.org/3/library/concurrent.futures.html) interface that allows for finer control and asynchronous computation.\n",
    "\n",
    "The `submit` function sends a function and arguments to the distributed scheduler for processing. They return `Future` objects that refer to remote data on the cluster. The `Future` returns immediately while the computations run remotely in the background. There is no blocking of the local Python session."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "\n",
    "def inc(x):\n",
    "    time.sleep(random.uniform(0, 2))\n",
    "    return x + 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "f = client.submit(inc, 7.2)  # Submits inc(7.2) to the distributed scheduler\n",
    "print(f)\n",
    "print(type(f))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once the computation for the `Future` is complete, you can retrieve the result using the `.result()` method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "f.result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `map` function can be used to apply a function on a sequence of arguments (similar to the built-in Python `map` function)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To delete `Futures` in distributed memory, use the `del` keyword"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "del f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = range(10)\n",
    "futures = client.map(inc, data)\n",
    "futures"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here a list of `Futures` are returned, one for each item in the sequence of arguments. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "futures"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = client.gather(futures)\n",
    "# Same as results = [future.result() for future in futures]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice what happens if we run the same calculation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = range(10)\n",
    "futures = client.map(inc, data)\n",
    "futures"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The results are ready right away ... and ... the keys are the same. That's because all of the same objects are involved, and the results are still in the cluster memory."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `concurrent.futures` API even allows you to submit tasks based on the output of other tasks. This gives more flexibility in situations where the computations may evolve over time."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```python\n",
    "from dask.distributed import as_completed\n",
    "\n",
    "seq = as_completed(futures)\n",
    "\n",
    "for future in seq:\n",
    "    y = future.result()\n",
    "    if condition(y):\n",
    "        new_future = client.submit(...)\n",
    "        seq.add(new_future)  # add back into the loop\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.close()"
   ]
  }
 ],
 "metadata": {
  "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.8.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}