Live Notebook

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

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

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.

[1]:

from dask.distributed import Client, progress
client

[1]:


### Cluster

• Workers: 1
• Cores: 4
• Memory: 8.36 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.

[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

time.sleep(random.random())
return x + y


We can run them locally

[3]:

inc(1)

[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.

[4]:

future = client.submit(inc, 1)  # returns immediately with pending future
future

[4]:

Future: inc status: pending, key: inc-c3b0b6f6072e9f86ab2ce44933897787

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

[5]:

future  # scheduler and client talk constantly

[5]:

Future: inc status: pending, key: inc-c3b0b6f6072e9f86ab2ce44933897787

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

[6]:

future.result()

[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.

[7]:

x = client.submit(inc, 1)
y = client.submit(double, 2)
z

[7]:

[8]:

z.result()

[8]:

6


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

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

[9]:

zs = []

[10]:

%%time

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)

CPU times: user 108 ms, sys: 24.6 ms, total: 132 ms
Wall time: 114 ms

[11]:

total = client.submit(sum, zs)


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)

[12]:

client.cluster.scale(10)  # ask for ten 4-thread workers


## 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

[13]:

L = zs
while len(L) > 1:
new_L = []
for i in range(0, len(L), 2):
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.

[14]:

del future, L, new_L, total  # clear out some old work

[15]:

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)