# Best Practices

It is easy to get started with Dask delayed, but using it well does require some experience. This page contains suggestions for best practices, and includes solutions to common problems.

## Call delayed on the function, not the result

Dask delayed operates on functions like `dask.delayed(f)(x, y)`, not on their results like `dask.delayed(f(x, y))`. When you do the latter, Python first calculates `f(x, y)` before Dask has a chance to step in.

Don’t | Do

    # This executes immediately

    dask.delayed(f(x, y))

    # This makes a delayed function, acting lazily

    dask.delayed(f)(x, y)

## Compute on lots of computation at once

To improve parallelism, you want to include lots of computation in each compute call. Ideally, you want to make many `dask.delayed` calls to define your computation and then call `dask.compute` only at the end. It is ok to call `dask.compute` in the middle of your computation as well, but everything will stop there as Dask computes those results before moving forward with your code.

Don’t | Do

    # Avoid calling compute repeatedly

    results = []
    for x in L:
        y = dask.delayed(f)(x)
        results.append(y.compute())

    results

    # Collect many calls for one compute

    results = []
    for x in L:
        y = dask.delayed(f)(x)
        results.append(y)

    results = dask.compute(*results)

Calling y.compute() within the loop would await the result of the computation every time, and so inhibit parallelism.

## Don’t mutate inputs

Your functions should not change the inputs directly.

Don’t | Do

    # Mutate inputs in functions

    @dask.delayed
    def f(x):
        x += 1
        return x

    # Return new values or copies

    @dask.delayed
    def f(x):
        x = x + 1
        return x

If you need to use a mutable operation, then make a copy within your function first:

    @dask.delayed
    def f(x):
        x = copy(x)
        x += 1
        return x

## Avoid global state

Ideally, your operations shouldn’t rely on global state. Using global state might work if you only use threads, but when you move to multiprocessing or distributed computing then you will likely encounter confusing errors.

Don’t

    L = []

    # This references global variable L

    @dask.delayed
    def f(x):
        L.append(x)

## Don’t rely on side effects

Delayed functions only do something if they are computed. You will always need to pass the output to something that eventually calls compute.

Don’t | Do

    # Forget to call compute

    dask.delayed(f)(1, 2, 3)

    ...

    # Ensure delayed tasks are computed

    x = dask.delayed(f)(1, 2, 3)
    ...
    dask.compute(x, ...)

In the first case here, nothing happens, because `compute()` is never called.

## Break up computations into many pieces

Every `dask.delayed` function call is a single operation from Dask’s perspective. You achieve parallelism by having many delayed calls, not by using only a single one: Dask will not look inside a function decorated with `@dask.delayed` and parallelize that code internally. To accomplish that, it needs your help to find good places to break up a computation.

Don’t | Do

    # One giant task

    def load(filename):
        ...

    def process(data):
        ...

    def save(data):
        ...

    @dask.delayed
    def f(filenames):
        results = []
        for filename in filenames:
            data = load(filename)
            data = process(data)
            result = save(data)
            results.append(result)

        return results

    dask.compute(f(filenames))

    # Break up into many tasks

    @dask.delayed
    def load(filename):
        ...

    @dask.delayed
    def process(data):
        ...

    @dask.delayed
    def save(data):
        ...

    def f(filenames):
        results = []
        for filename in filenames:
            data = load(filename)
            data = process(data)
            result = save(data)
            results.append(result)

        return results

    dask.compute(f(filenames))

The first version only has one delayed task, and so cannot parallelize.

## Avoid too many tasks

Every delayed task has an overhead of a few hundred microseconds. Usually this is ok, but it can become a problem if you apply `dask.delayed` too finely. In this case, it’s often best to break up your many tasks into batches or use one of the Dask collections to help you.

Don’t | Do

    # Too many tasks

    results = []
    for x in range(10000000):
        y = dask.delayed(f)(x)
        results.append(y)

    # Use collections

    import dask.bag as db
    b = db.from_sequence(range(10000000), npartitions=1000)
    b = b.map(f)
    ...

Here we use `dask.bag` to automatically batch applying our function. We could also have constructed our own batching as follows

    def batch(seq):
        sub_results = []
        for x in seq:
            sub_results.append(f(x))
        return sub_results

     batches = []
     for i in range(0, 10000000, 10000):
         result_batch = dask.delayed(batch)(range(i, i + 10000))
         batches.append(result_batch)

Here we construct batches where each delayed function call computes for many data points from the original input.

## Avoid calling delayed within delayed functions

Often, if you are new to using Dask delayed, you place `dask.delayed` calls everywhere and hope for the best. While this may actually work, it’s usually slow and results in hard-to-understand solutions.

Usually you never call `dask.delayed` within `dask.delayed` functions.

Don’t | Do

    # Delayed function calls delayed

    @dask.delayed
    def process_all(L):
        result = []
        for x in L:
            y = dask.delayed(f)(x)
            result.append(y)
        return result

    # Normal function calls delayed

    def process_all(L):
        result = []
        for x in L:
            y = dask.delayed(f)(x)
            result.append(y)
        return result

Because the normal function only does delayed work it is very fast and so there is no reason to delay it.

## Don’t call dask.delayed on other Dask collections

When you place a Dask array or Dask DataFrame into a delayed call, that function will receive the NumPy or Pandas equivalent. Beware that if your array is large, then this might crash your workers.

Instead, it’s more common to use methods like `da.map_blocks`

Don’t | Do

    # Call delayed functions on Dask collections

    import dask.dataframe as dd
    df = dd.read_csv('/path/to/*.csv')

    dask.delayed(train)(df)

    # Use mapping methods if applicable

    import dask.dataframe as dd
    df = dd.read_csv('/path/to/*.csv')

    df.map_partitions(train)

Alternatively, if the procedure doesn’t fit into a mapping, you can always turn your arrays or dataframes into many delayed objects, for example

    partitions = df.to_delayed()
    delayed_values = [dask.delayed(train)(part)
                      for part in partitions]

However, if you don’t mind turning your Dask array/DataFrame into a single chunk, then this is ok.

    dask.delayed(train)(..., y=df.sum())

## Avoid repeatedly putting large inputs into delayed calls

Every time you pass a concrete result (anything that isn’t delayed) Dask will hash it by default to give it a name. This is fairly fast (around 500 MB/s) but can be slow if you do it over and over again. Instead, it is better to delay your data as well.

This is especially important when using a distributed cluster to avoid sending your data separately for each function call.

Don’t | Do

    x = np.array(...)  # some large array

    results = [dask.delayed(train)(x, i)
               for i in range(1000)]

    x = np.array(...)    # some large array
    x = dask.delayed(x)  # delay the data once
    results = [dask.delayed(train)(x, i)
               for i in range(1000)]

Every call to `dask.delayed(train)(x, ...)` has to hash the NumPy array `x`, which slows things down.