Styjun

I have a pipeline that transforms (maps) a dataframe. The output is large - rows in the input dataframe contain audio in binary format and rows in the output dataframe contain extracted binary features.

I'm reading the input from a partitioned parquet file and writing it back to a different parquet file(s) - both on a network share.

From my understanding, in distributed dask, each worker will send the output back to the scheduler (and then maybe the scheduler sends it back to the client??) and only then will the scheduler (or the client) write it to the network share. Is this correct?

If yes, if the data is big and bandwidth is an issue it seems there is redundant communication in this scenario - why can't the workers send the output directly to the final destination (network share in this case)? Certainly, the share needs to be available to all workers, and someone needs to synchronize the writes, but isn't this what the magic of dask is about?

Your understanding is incorrect: the workers will read and write to shared storage or cloud/network services directly, this is the normal way that things are calculated.

df = dd.read_parquet(url)
df_out = do_work(df)
df_out.to_parquet(url2)

In this case, the data is never seen by the scheduler or the client. They do communicate, though: the client will load metadata about the dataset, so that it can make inferences about how to split up the work to be done, and the scheduler talks to both the client and the workers to farm out these task specifications and check when they are done.

You can optionally bring the whole dataset into the memory of the client as a pandas dataframe with

local_df = df.compute()

but this is optional and obviously not recommended where the data size is bigger than memory. You usually never need to do this for the whole dataset, only maybe for some aggregate result much smaller than the original. Even in this case, the scheduler itself does not store the results.