Dask diagnostics - progress bar with map_partition / delayed

user_605328 0
Viewed 664

I am using the distributed scheduler and distributed progressbar. Is there a way of having the progress bar work for Dataframe.map_partition or delayed? I assume the lack of futures is what causes the bar not to work. If I change my code to client.submit the progressbar does work.

  • Code looks like this:
import dask.dataframe as dd
from distributed import Client
from distributed.diagnostics.progressbar import progress

client = Client("tcp://....")
...
ddf = dd.read_parquet("...")
ddf = ddf.map_partitions(..)
progress(ddf)  # no futures to pass
dask.compute(ddf)
  • Alternative with dask.delayed does not work either:
delayed = [dask.deplayed(myfunc)(ddf.get_partition(i)) for i in range(ddf.npartitions)]
progress(delayed)
dask.compute(*delayed)
  • Client.submit does produce a working progress bar, but code execution fails and I haven't managed to debug it yet.
futures = [client.submit(myfunc, ddf.get_partition(i)) for i in range(ddf.npartitions)]
progress(futures)
dask.compute(*futures)

Is there a way to get the progress bar (or a report of tasks completed vs total) working for map_partitions or dask.delayed ?

  • Full code example with delayed:
import dask
import npumpy as np
import pandas as pd
import dask.dataframe as dd
from distributed import Client
from distributed.diagnostics.progressbar import progress
import time

cl = Client("tcp://10.0.2.15:8786")

def wait(df):
    print("Received chunk")
    time.sleep(2)
    print("finish")

df = pd.DataFrame(np.random.randint(0, 100, size=(1000, 4)), columns=list('ABCD'))
ddf = dd.from_pandas(df, npartitions=4)

futures = [dask.delayed(wait)(ddf.get_partition(i)) for i in range(ddf.npartitions)]
progress(futures)
dask.compute(*futures)
dask-distributed dask pandas numpy python
1 Answers

Yes you are right, progress is intended to work with futures or collections that contain futures. You don't need submit a big list of futures to use it, though:

ddf = ddf.map_partitions(..)
fut = client.compute(ddf)
progress(fut)
# wait on fut, call fut.result() or continue

Also don't forget: the distributed scheduler that you are using, even if on a single machine only, comes with a diagnostics dashboard that contains the same information. Usually this is at http://localhost:8787, and you can access from any browser.

user_38211540
Related