解压元组列表的 dask 延迟对象

Question

我有一个返回两个元素的元组的函数。使用 pool starmap 调用该函数来生成元组列表，这些元组被解压缩为两个列表。

def func():
   #...some operations
   return (x,y)

def MP_a_func(func,iterable,proc,chunk):
    pool=multiprocessing.Pool(processes=proc)
    Result=pool.starmap(func,iterable,chunksize=chunk)
    pool.close()
    return Result
##
if __name__ == '__main__':
    results=MP_a_func(func,iterable,proc,chunk)

a,b=zip(*results)

我现在希望使用 dask

delayed

API，如下所示

if __name__ == '__main__':
    results=delayed(MP_a_func(func,iterable,proc,chunk))

是否可以在不使用

results.compute()

的情况下解压延迟对象中的元组？

感谢您的帮助

Answer 1

另一个延迟函数可以解压元组，在下面的示例中，

return_tuple(1)

的延迟值没有被计算，而是作为

delayed

对象传递：

import dask

@dask.delayed
def return_tuple(x):
    return x+1, x-1

@dask.delayed
def process_first_item(some_tuple):
    return some_tuple[0]+10

result = process_first_item(return_tuple(1))

dask.compute(result)

根据@mdurant的回答，事实证明

delayed

函数/装饰器有

nout

参数，另请参阅这个答案。

Answer 2

如果您知道输出的数量，则

delayed

函数（或装饰器）采用可选的

nout

参数，这会将单个延迟拆分为多个延迟输出。这听起来正是您所需要的。