distributed: Workers stuck, increased memory usage while processing large CSV from S3.

I’m processing a dataframe stored as a (relatively) large CSV on S3. Using distributed scheduler with multiprocessing (1 thread per 1 worker process, --no-nanny). Workers seem to be accumulating data and getting stuck, in some cases this also leads to failure of whole job.

I came up with minimal reproducing example as below (only read/write CSV)

frame = df.read_csv(input_url,
                    collection=True,
                    blocksize=1024*1024,
                    compression=None,
                    lineterminator='\n',
                    dtype=str,
                    sep=',',
                    quotechar='"',
                    encoding='utf-8')

fun_list = frame.to_csv(output_url,
                        compute=False,
                        encoding='utf-8',
                        index=False,
                        index_label=False)

futures = client.compute(fun_list)
progress(futures)
client.gather(futures)

This would hang forever with progress at 0%. In worker log: distributed.worker - WARNING - Memory use is high but worker has no data to store to disk. Perhaps some other process is leaking memory? Process memory: 7.16 GB -- Worker memory limit: 8.02 GB

The file itself is only 1.2GB though. Using distributed 1.19.2 and dask 0.15.4