dask - 我如何迭代 Dask 任务图中的集合

Question

我有以下任务图。

def getdocs():
    return [doc1,doc2]

def tosents(doc):
    return doc.split(".")  # Except doc is a list of docs

def towords(sent):
    return sent.split(" ")  

graph = {
    "documents": (getdocs,),
    "sentences": (tosents,"documents"),
    "words": (towords, "sentences")
}

每个任务都返回一个列表，但我希望将这些结果并行化到接收函数，该函数应该只接受一个值（例如字符串）。目前，结果正在传递中。

如何指示结果将通过接收函数进行枚举？使用产量？

提前谢谢你。

标签： dask

Answer 1

人们构建自己的 Dask 图是不寻常的，因为该delayed界面非常易于使用。如果您有一个返回已知数量的输出的函数，您打算单独处理这些输出（即并行操作），您可以执行类似的操作

ddocs = delayed(getdocs, nout=2)
sentences = [delayed(tosentences)(doc) for doc in ddocs()]

wheretosentences需要一个文档作为输入。的每个元素都sentences将是一个带有自己的图形的延迟对象。如果将它们一起传递给compute().

-编辑-

ddocs需要调用；你不能迭代延迟函数，但你可以迭代它产生的延迟对象。