Dask 异步处理，在结果到达时打印结果

Question

以下示例不起作用，除非 asynchronous 关键字未在 Localcluster 中使用。我想控制使用了多少 processes/workers 并并行处理函数，并在结果准备就绪时打印出来。需要更改什么？

import time
from dask.distributed import Client, LocalCluster, as_completed


def wait(sec):
    time.sleep(sec)
    return sec


def main():

    cluster = LocalCluster(n_workers=2, ncores=2, asynchronous=True)
    inputs = [5, 7, 3, 1]
    client = Client(cluster)
    futures = client.map(wait, inputs)

    for future, result in as_completed(futures, with_results=True):
        print(result)

    client.close()


if __name__ == '__main__':
    main()

Answer 1

按照您的建议，您应该从 LocalCluster 调用中删除 asynchronous= 关键字。该关键字用于支持异步函数，如下所示：

async def main():
    cluster = await LocalCluster(n_workers=2, ncores=2, asynchronous=True)
    inputs = [5, 7, 3, 1]
    client = await Client(cluster, asynchronous=True)
    futures = client.map(wait, inputs)

    async for future, result in as_completed(futures, with_results=True):
        print(result)

    await client.close()

如果您不想使用 async-await 语法（这种情况比较少见），那么您应该忽略 asynchronous= 关键字。它可能并不像你想象的那样。

Dask 异步处理，在结果到达时打印结果

Dask async processing, print results as they arrive

python

asynchronous

cluster-computing

dask