python - 将 dask delayed 与函数返回列表一起使用

Question

我正在尝试使用 dask.delayed 来构建任务图。这大部分工作得很好，但我经常遇到这样的情况，在这种情况下，我有许多延迟对象，这些对象有一个方法返回一个对象列表，其长度不容易根据我目前可用的信息计算得出:

items = get_collection() # known length

def do_work(item):
    # get_list_of_things returns list of "unknown" length
    return map(lambda x: x.DoStuff(), item.get_list_of_things())

results = [delayed(do_work(x)) for x in items]

这给出了一个

TypeError: Delayed objects of unspecified length are not iterable

dask 有什么方法可以解决这个问题，最好是不必在中间结果上调用 .compute()，因为那样会破坏拥有任务图的大部分好处？它基本上意味着在运行某些步骤之前无法完全解析图形，但唯一可变的是平行部分的宽度，它不会改变图形的结构或深度。

Answer 1

不幸的是，如果您想对列表中的每个元素调用一个单独的函数，那么是您的图形结构的一部分，并且如果您想在图形构建时必须知道使用 dask.delayed。

一般来说，我看到两个选项:

1. 不要为列表中的每个元素分别创建一个任务，而是为前 10%、第二个 10% 等创建一个任务。这与 dask.bag 中采用的方法相同，它还处理具有未知数量元素的并行性(这可能值得考虑。

   http://dask.pydata.org/en/latest/bag.html

2. 切换到实时 concurrent.futures 界面，等待您的列表结果，然后再提交更多工作

   from dask.distributed import Client
   client = Client()
   list_future = client.submit(do_work, *args)
   len_future = client.submit(len, list_future)
   
   n = len_future.result()  # wait until the length is computed
   
   futures = [client.submit(operator.getitem, list_future, i) for i in range(n)]
   
   ... do more stuff with futures
   

   http://dask.pydata.org/en/latest/futures.html