dask.compute()函数怎么使用scheduler参数指定distributed调度器进行并行计算

在使用dask.compute()函数时，可以通过设置scheduler参数指定distributed调度器进行并行计算。具体的语法如下：

dask.compute(*args, **kwargs, scheduler='distributed')

其中，args和kwargs是要计算的dask对象，scheduler参数指定为'distributed'即可使用distributed调度器。例如，下面的代码演示了如何使用distributed调度器计算一个dask数组：

import dask.array as da
from dask.distributed import Client

client = Client()  # 连接到分布式集群

x = da.random.uniform(size=(1000, 1000), chunks=(100, 100))
y = (x + x.T) - x.mean(axis=0)

result = dask.compute(y, scheduler='distributed')  # 使用distributed调度器计算结果

print(result)

需要注意的是，在使用distributed调度器时，需要先连接到一个分布式集群。上面的代码中，我们使用了dask.distributed.Client()对象连接到一个分布式集群，并通过client对象来执行计算。

相关问题

dask.compute()函数怎么使用distributed调度器进行并行计算？

`dask.compute()`函数是用于触发Dask图计算的方法。如果要使用Distributed调度器进行并行计算，可以按照以下步骤：

1. 创建一个`Client`对象，指定Distributed调度器的地址：

from dask.distributed import Client
client = Client('scheduler_address:8786')

2. 构建Dask图，并使用`client.submit()`方法将任务提交到Distributed调度器上：

from dask import delayed

@delayed
def inc(x):
    return x + 1

@delayed
def double(x):
    return x * 2

data = [1, 2, 3, 4, 5]
incs = [inc(i) for i in data]
dbls = [double(i) for i in incs]
results = sum(dbls)
futures = client.compute(results)

3. 使用`dask.distributed.wait()`等待所有任务完成：

from dask.distributed import wait
wait(futures)

4. 使用`dask.distributed.gather()`方法收集任务结果：

from dask.distributed import gather
results = gather(futures)

5. 关闭`Client`对象：

client.close()

这些步骤可以确保Dask图在Distributed调度器上进行并行计算。