You can run this notebook in a live session Binder or view it on Github.

Custom Workloads with Futures

Dask futures provide fine-grained real-time execution for custom situations. This is the foundation for other APIs like Dask arrays and dataframes.

Start Dask Client

Unlike for arrays and dataframes, you need the Dask client to use the Futures interface. Additionally the client provides a dashboard which is useful to gain insight on the computation.

The link to the dashboard will become visible when you create the client below. We recommend having it open on one side of your screen while using your notebook on the other side. This can take some effort to arrange your windows, but seeing them both at the same is very useful when learning.

In [1]:
from dask.distributed import Client, progress
client = Client(threads_per_worker=4, n_workers=1)
client
Out[1]:

Client

Cluster

  • Workers: 1
  • Cores: 4
  • Memory: 7.84 GB

Create simple functions

These functions do simple operations like add two numbers together, but they sleep for a random amount of time to simulate real work.

In [2]:
import time
import random

def inc(x):
    time.sleep(random.random())
    return x + 1

def double(x):
    time.sleep(random.random())
    return 2 * x

def add(x, y):
    time.sleep(random.random())
    return x + y

We can run them locally

In [3]:
inc(1)
Out[3]:
2

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.

In [4]:
future = client.submit(inc, 1)  # returns immediately with pending future
future
Out[4]:
Future: inc status: pending, key: inc-efb82fe5387f793c9eceecdf18420ff3

If you wait a second, and then check on the future again, you’ll see that it has finished.

In [5]:
future  # scheduler and client talk constantly
Out[5]:
Future: inc status: pending, key: inc-efb82fe5387f793c9eceecdf18420ff3

You can block on the computation and gather the result with the .result() method.

In [6]:
future.result()
Out[6]:
2

Chain dependencies

You can submit tasks on other futures. This will create a dependency between the inputs and outputs. Dask will track the execution of all tasks, ensuring that downstream tasks are run at the proper time and place and with the proper data.

In [7]:
x = client.submit(inc, 1)
y = client.submit(double, 2)
z = client.submit(add, x, y)
z
Out[7]:
Future: add status: pending, key: add-a0180be60f687367c2129dda7417bb28
In [8]:
z.result()
Out[8]:
6

Note that we never blocked on x or y nor did we ever have to move their data back to our notebook.

Submit many tasks

So we’ve learned how to run Python functions remotely. This becomes useful when we add two things:

  1. We can submit thousands of tasks per second
  2. Tasks can depend on each other by consuming futures as inputs

We submit many tasks that depend on each other in a normal Python for loop

In [9]:
%%time
zs = []
for i in range(256):
    x = client.submit(inc, i)     # x = inc(i)
    y = client.submit(double, x)  # y = inc(x)
    z = client.submit(add, x, y)  # z = inc(y)
    zs.append(z)

total = client.submit(sum, zs)
CPU times: user 156 ms, sys: 32 ms, total: 188 ms
Wall time: 172 ms

To make this go faster, add an additional workers with more cores

(although we’re still only working on our local machine, this is more practical when using an actual cluster)

In [10]:
for i in range(10):
    client.cluster.start_worker(ncores=4)

Custom computation: Tree summation

As an example of a non-trivial algorithm, consider the classic tree reduction. We accomplish this with a nested for loop and a bit of normal Python logic.

finish           total             single output
    ^          /        \
    |        c1          c2        neighbors merge
    |       /  \        /  \
    |     b1    b2    b3    b4     neighbors merge
    ^    / \   / \   / \   / \
start   a1 a2 a3 a4 a5 a6 a7 a8    many inputs
In [11]:
L = zs
while len(L) > 1:
    new_L = []
    for i in range(0, len(L), 2):
        future = client.submit(add, L[i], L[i + 1])  # add neighbors
        new_L.append(future)
    L = new_L                                   # swap old list for new

If you’re watching the dashboard’s status page then you may want to note two things:

  1. The red bars are for inter-worker communication. They happen as different workers need to combine their intermediate values
  2. There is lots of parallelism at the beginning but less towards the end as we reach the top of the tree where there is less work to do.

Alternatively you may want to navigate to the dashboard’s graph page and then run the cell above again. You will be able to see the task graph evolve during the computation.

Building a computation dynamically

In the examples above we explicitly specify the task graph ahead of time. We know for example that the first two futures in the list L will be added together.

Sometimes this isn’t always best though, sometimes you want to dynamically define a computation as it is happening. For example we might want to sum up these values based on whichever futures show up first, rather than the order in which they were placed in the list to start with.

For this, we can use operations like as_completed.

We recommend watching the dashboard’s graph page when running this computation. You should see the graph construct itself during execution.

In [12]:
del future, L, new_L, total  # clear out some old work
In [13]:
from dask.distributed import as_completed

zs = client.map(inc, zs)
seq = as_completed(zs)

while seq.count() > 2:  # at least two futures left
    a = next(seq)
    b = next(seq)
    new = client.submit(add, a, b)  # add them together
    seq.add(new)                    # add new future back into loop
In [14]: