python - 如何跨多个调用将 dask-DAG 持久保存在分布式集群上并保留中间结果？

Question

我尝试在分布式客户端的多次调用中提交dask-DAG，但无法在集群上保留中间结果。你能指出，我该如何解决这个问题吗？

from distributed import Client
c = Client()


dsk0 = {'a': 1, 'b': (lambda x: 2*x, 'a')}
keys0 = ['a', 'b']
futures0 = c._graph_to_futures(dsk0, keys0)
fb = futures0['b']
b = fb.result()  # Yields correctly 2

dsk1 = {'c': (lambda x: 3*x, 'a')}
keys1 = ['c']
futures1 = c._graph_to_futures(dsk1, keys1)
fc = futures1['c']
c = fc.result()  # Yields 'aaa', instead of 3

提前致谢!

马库斯

Answer 1

我建议使用dask.delayed和 client.compute方法

from dask import delayed
from distributed import Client
client = Client()

a = delayed(1)
b = delayed(lambda x: 2 * x)(a)

a_future, b_future = client.compute([a, b])

>>> b_future.result()
2

c = delayed(lambda x: 3 * x)(a_future)
c_future = client.compute(c)

>>> c_future.result()
3

像 _graph_to_futures 这样直接处理图形的内部函数更容易出错，通常供内部使用。