TopolearnMemory pressure causes most failed Dask jobs, and persist plus the dashboard are how you control and see it.
df = df.persist() starts computing a collection on the cluster and keeps the results in worker memory, returning a collection backed by those in-memory pieces. Use it when you will reuse the same intermediate data many times, as in interactive exploration or iterative algorithms. compute() instead brings the result back to your process.
Persist only what fits comfortably in total cluster memory. Each worker has a memory_limit and reacts in stages: around 60% of the limit it spills data to disk, around 70% it also spills based on process memory, at 80% it pauses accepting new tasks, and at 95% the nanny restarts the worker, losing its data. Frequent spilling or restarts mean partitions are too large, too much is persisted, or a task creates huge intermediate objects.
The dashboard (client.dashboard_link) shows this live:
- Task stream: what runs where; red is communication, white is idle time.
- Worker memory bars: grey for managed memory, orange for spilled, with the limit marked.
- Progress, graph and profile pages to find slow functions.
client.get_task_stream() and performance_report(filename=...) save diagnostics for later. Use wait(df) to block until a persist finishes, and del df or client.cancel(df) to release memory.
Example
import dask.dataframe as dd
import pandas as pd
from dask.distributed import Client, performance_report, wait
if __name__ == "__main__":
with Client(n_workers=2, threads_per_worker=1, memory_limit="1GB") as client:
pdf = pd.DataFrame({"k": [i % 10 for i in range(100_000)],
"v": range(100_000)})
df = dd.from_pandas(pdf, npartitions=10)
clean = df[df.v % 3 == 0].persist() # computed once, kept on workers
wait(clean)
mem = client.run(lambda dask_worker: len(dask_worker.data))
print(sum(mem.values())) # 10 (partitions held in memory)
with performance_report(filename="report.html"): # needs bokeh
print(clean.v.sum().compute()) # 1666683333
print(clean.groupby("k").v.mean().compute().round().iloc[0]) # 49995.0
del clean # frees worker memoryRemember
persist()keeps computed partitions on workers for reuse- Workers spill, pause and restart at rising memory thresholds
- The dashboard's task stream and memory bars explain most problems
Watch out for
- Persisting a dataset larger than cluster memory and causing constant spilling or worker restarts.