如何系统地重用 Dask 中延迟函数的结果？

Question

我正在使用 Dask 构建计算图。一些中间值将被多次使用，但我希望这些计算只运行一次。我一定是犯了一个小错误，因为事实并非如此。这是一个最小的例子：

In [1]:    import dask
           dask.__version__
    
Out [1]:   '1.0.0'

In [2]:   class SumGenerator(object):
              def __init__(self):
                  self.sources = []
    
              def register(self, source):
                  self.sources += [source]
        
              def generate(self):
                  return dask.delayed(sum)([s() for s in self.sources])

In [3]:    sg = SumGenerator()

In [4]:    @dask.delayed
           def source1():
               return 1.

           @dask.delayed
           def source2():
               return 2.

           @dask.delayed
           def source3():
               return 3.

In [5]:    sg.register(source1)
           sg.register(source1)
           sg.register(source2)
           sg.register(source3)

In [6]:    sg.generate().visualize()

遗憾的是我无法 post 生成的图形图像，但基本上我看到函数 source1 注册了两次的两个独立节点。因此该函数被调用了两次。我宁愿让它被调用一次，结果被记住并在总和中加了两次。这样做的正确方法是什么？

Answer 1

您需要通过传递 pure=True 参数来调用 dask.delayed 装饰器。

来自dask delayed docs

delayed also accepts an optional keyword pure. If False, then subsequent calls will always produce a different Delayed

If you know a function is pure (output only depends on the input, with no global state), then you can set pure=True.

所以使用那个

import dask

class SumGenerator(object):
    def __init__(self):
        self.sources = []

    def register(self, source):
        self.sources += [source]

    def generate(self):
        return dask.delayed(sum)([s() for s in self.sources])

@dask.delayed(pure=True)
def source1():
    return 1.

@dask.delayed(pure=True)
def source2():
    return 2.

@dask.delayed(pure=True)
def source3():
    return 3.

sg = SumGenerator()

sg.register(source1)
sg.register(source1)
sg.register(source2)
sg.register(source3)

sg.generate().visualize()

输出和图表

使用 print(dask.compute(sg.generate())) 得到 (7.0,) 与您编写的相同，但没有图中所示的额外节点。

如何系统地重用 Dask 中延迟函数的结果？

How can I systematically reuse the results of delayed functions in Dask?

python

dask

dask-delayed