Caching when using a distributed scheduler

[xinrong-meng]

Let's say we have a large Dask DataFrame df on a distributed cluster, and we filter df using filter_expr.

And then we have a bunch of operations on df[filter_expr], such as count, mean and groupby, etc.

Here, calling df[filter_expr].compute() to cache doesn't work because df[filter_expr] doesn't fit into memory.

Opportunistic caching also doesn't seem to work because we are using a distributed scheduler.

In this case, what would be the best caching mechanism?

[andersy005]

@xinrong-databricks,

Here, calling df[filter_expr].compute() to cache doesn't work because df[filter_expr] doesn't fit into memory.

have you tried using .persist() i.e. df[filter_expr].persist() to persist your results in the your cluster's distributed memory?

[xinrong-meng]

Thank you, @andersy005!

May I take it as compute() uses scheduler's memory only?

Is there a function that blocks until the computation is finished (like compute()) and utilizes the cluster's distributed memory(like persist())?

[andersy005]

May I take it as compute() uses scheduler's memory only?

With .compute(), the data/results are moved to wherever the client resides.

Is there a function that blocks until the computation is finished (like compute()) and utilizes the cluster's distributed memory(like persist())?

there's a wait() function for this purpose:

In [5]: from distributed import Client, wait

In [6]: client = Client()

In [9]: df = client.persist(df)

In [10]: wait(df)

[xinrong-meng]

Thank you @andersy005, that's helpful!