Managing Computation¶

Data and Computation in Dask.distributed are always in one of three states

Concrete values in local memory. Example include the integer 1 or a numpy array in the local process.
Lazy computations in a dask graph, perhaps stored in a dask.delayed or dask.dataframe object.
Running computations or remote data, represented by Future objects pointing to computations currently in flight.

All three of these forms are important and there are functions that convert between all three states.

Dask Collections to Concrete Values¶

You can turn any dask collection into a concrete value by calling the .compute() method or dask.compute(...) function. This function will block until the computation is finished, going straight from a lazy dask collection to a concrete value in local memory.

This approach is the most familiar and straightforward, especially for people coming from the standard single-machine Dask experience or from just normal programming. It is great when you have data already in memory and want to get small fast results right to your local process.

>>> df = dd.read_csv('s3://...')
>>> df.value.sum().compute()
100000000

However, this approach often breaks down if you try to bring the entire dataset back to local RAM

>>> df.compute()
MemoryError(...)

It also forces you to wait until the computation finishes before handing back control of the interpreter.

Dask Collections to Futures¶

You can asynchronously submit lazy dask graphs to run on the cluster with the client.compute and client.persist methods. These functions return Future objects immediately. These futures can then be queried to determine the state of the computation.

client.compute¶

The .compute method takes a collection and returns a single future.

>>> df = dd.read_csv('s3://...')
>>> total = client.compute(df.sum())  # Return a single future
>>> total
Future(..., status='pending')

>>> total.result()               # Block until finished
100000000

Because this is a single future the result must fit on a single worker machine. Like dask.compute above, the client.compute method is only appropriate when results are small and should fit in memory. The following would likely fail:

>>> future = client.compute(df)       # Blows up memory

Instead, you should use client.persist

client.persist¶

The .persist method submits the task graph behind the Dask collection to the scheduler, obtaining Futures for all of the top-most tasks (for example one Future for each Pandas DataFrame in a Dask DataFrame). It then returns a copy of the collection pointing to these futures instead of the previous graph. This new collection is semantically equivalent but now points to actively running data rather than a lazy graph. If you look at the dask graph within the collection you will see the Future objects directly:

>>> df = dd.read_csv('s3://...')
>>> df.dask                          # Recipe to compute df in chunks
{('read', 0): (load_s3_bytes, ...),
 ('parse', 0): (pd.read_csv, ('read', 0)),
 ('read', 1): (load_s3_bytes, ...),
 ('parse', 1): (pd.read_csv, ('read', 1)),
 ...
}

>>> df = client.persist(df)               # Start computation
>>> df.dask                          # Now points to running futures
{('parse', 0): Future(..., status='finished'),
 ('parse', 1): Future(..., status='pending'),
 ...
}

The collection is returned immediately and the computation happens in the background on the cluster. Eventually all of the futures of this collection will be completed at which point further queries on this collection will likely be very fast.

Typically the workflow is to define a computation with a tool like dask.dataframe or dask.delayed until a point where you have a nice dataset to work from, then persist that collection to the cluster and then perform many fast queries off of the resulting collection.

Concrete Values to Futures¶

We obtain futures through a few different ways. One is the mechanism above, by wrapping Futures within Dask collections. Another is by submitting data or tasks directly to the cluster with client.scatter, client.submit or client.map.

futures = client.scatter(args)                        # Send data
future = client.submit(function, *args, **kwargs)     # Send single task
futures = client.map(function, sequence, **kwargs)    # Send many tasks

In this case *args or **kwargs can be normal Python objects, like 1 or 'hello', or they can be other Future objects if you want to link tasks together with dependencies.

Unlike Dask collections like dask.delayed these task submissions happen immediately. The concurrent.futures interface is very similar to dask.delayed except that execution is immediate rather than lazy.

Futures to Concrete Values¶

You can turn an individual Future into a concrete value in the local process by calling the Future.result() method. You can convert a collection of futures into concrete values by calling the client.gather method.

>>> future.result()
1

>>> client.gather(futures)
[1, 2, 3, 4, ...]

Futures to Dask Collections¶

As seen in the Collection to futures section it is common to have currently computing Future objects within Dask graphs. This lets us build further computations on top of currently running computations. This is most often done with dask.delayed workflows on custom computations:

>>> x = delayed(sum)(futures)
>>> y = delayed(product)(futures)
>>> future = client.compute(x + y)

Mixing the two forms allow you to build and submit a computation in stages like sum(...) + product(...). This is often valuable if you want to wait to see the values of certain parts of the computation before determining how to proceed. Submitting many computations at once allows the scheduler to be slightly more intelligent when determining what gets run.

If this page interests you then you may also want to check out the doc page on Managing Memory