3

我正在尝试使用 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() ，因为这会破坏拥有任务图的大部分好处？它基本上意味着在运行某些步骤之前无法完全解析图形，但唯一可变的是平行部分的宽度，它不会改变图形的结构或深度。

4

1 回答 1

3

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

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

  1. 不要为列表中的每个元素创建一个单独的任务，而是为前 10%、后 10% 等创建一个任务。这与 dask.bag 中采用的方法相同，它也使用未知数量的元素（这可能值得考虑。

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

  2. 切换到实时concurrent.futures界面，等待你的list结果再提交更多工作

    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

于 2017-12-12T01:33:31.523 回答