Contents

Custom Workloads with Futures

Live Notebook

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

Custom Workloads with Futures

Dask logo

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.

[1]:

from dask.distributed import Client, progress
client = Client(threads_per_worker=4, n_workers=1)
client

[1]:

Client

Client-fee34b1a-0ddf-11ed-9a76-000d3a8f7959

Connection method: Cluster object Cluster type: distributed.LocalCluster
Dashboard: http://127.0.0.1:8787/status

Cluster Info

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

def add(x, y):
    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, type: NoneType, key: inc-6a39f735d591f959f1c67a637717a113

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, type: NoneType, key: inc-6a39f735d591f959f1c67a637717a113

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 = client.submit(add, x, y)
z

[7]:
Future: add status: pending, type: NoneType, key: add-a255b15196528b8a69e0ea73bff2be61
[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.

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

[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 2.58 s, sys: 92.2 ms, total: 2.68 s
Wall time: 2.57 s

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

[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() > 1:  # at least two futures left
    a = next(seq)
    b = next(seq)
    new = client.submit(add, a, b, priority=1)  # add them together
    seq.add(new)                                # add new future back into loop

previous

Custom Workloads with Dask Delayed

next

Dask for Machine Learning