Logs are shown immediately after submitting a job via client on using dask

userBKzS 0
Viewed 29

The logs of the function submitted via the client are immediately displayed. Instead, the logs are expected to be displayed on client.gather(futures). The expected behavior could be achieved using Delayed but not using Futures.

Here is the code to reproduce issue:

from dask.distributed import Client

client = Client(processes=False, n_workers=2)

def inc(x):
    warning(f"{x}")
    return x + 1

output=[]
for x in [1, 2, 3, 4, 5]:
    a = client.submit(inc, x)
    output.append(a)

The above-added code will already display the logs on submission as shown below.

Output:

2022-09-19 20:55:23 ⚡ [root] 1
2022-09-19 20:55:23 ⚡ [root] 2
2022-09-19 20:55:23 ⚡ [root] 3
2022-09-19 20:55:23 ⚡ [root] 4
2022-09-19 20:55:23 ⚡ [root] 5

Output of client.gather(output)

[2, 3, 4, 5, 6]

But it is expected to show only at the execution of client.gather(output) along with the return of the results.

Intended behavior using Dask Delayed:

import dask

@dask.delayed
def inc(x):
    warning(f"{x}")
    return x + 1

data = [1, 2, 3, 4, 5]
output = []
for x in data:
    a = inc(x)
    output.append(a)

total = dask.delayed(output)

total.compute()

Output:

2022-09-19 21:05:01 ⚡ [root] 3
2022-09-19 21:05:01 ⚡ [root] 1
2022-09-19 21:05:01 ⚡ [root] 4
2022-09-19 21:05:01 ⚡ [root] 5
2022-09-19 21:05:01 ⚡ [root] 2
[2, 3, 4, 5, 6]

Could we get the expected behavior using the dask futures?

dask-delayed dask-distributed dask scheduler python
1 Answers

your assumption is not correct. Dask futures is a wraper around python base module concurrent.futures.

From the Futures documentation (emphasis added):

This interface is good for arbitrary task scheduling like dask.delayed, but is immediate rather than lazy, which provides some more flexibility in situations where the computations may evolve over time. These features depend on the second generation task scheduler found in dask.distributed (which, despite its name, runs very well on a single machine).

What this basically means is that task computations start exactly at the time they are submitted to the client (not the case for dask delayed). In your case you are storing the futures in a list which forces dask to keep the result of that particular future in memory once is computed and thus, you can call gather on that future to recover them from distributed memory.

As an example, in your case you have 2 workers if you introduce a small delay in your function, you will see that two elements will be printed at the same time once the workers are free the next two tasks will be submitted

import dask
import time

def inc(x):
    warning(f"{x}")
    time.sleep(2)
    return x + 1

output=[]
for x in [1, 2, 3, 4, 5]:
    a = client.submit(inc, x)
    output.append(a)

the output should look something like this

2022-09-19 20:55:23 ⚡ [root] 1
2022-09-19 20:55:23 ⚡ [root] 2
2022-09-19 20:57:23 ⚡ [root] 3
2022-09-19 20:57:23 ⚡ [root] 4
2022-09-19 20:59:23 ⚡ [root] 5
Lucas M. Uriarte0
Related