# Managing Computation

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

1. Concrete values in local memory. Example include the integer `1` or a numpy array in the local process.

2. Lazy computations in a dask graph, perhaps stored in a `dask.delayed` or `dask.dataframe` object.

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