用Xarray -Compute（）结果延迟的dask仍然延迟

Question

我试图在两个数据集上使用Dask和Xarray进行一些分析（例如AVG），然后计算两个结果之间的差异。

这是我的代码，

cluster = LocalCluster(n_workers=5, threads_per_worker=3, **worker_kwargs)

def calc_avg(path):
    
    mean = xr.open_mfdataset( path,combine='nested', concat_dim="time", parallel=True, decode_times=False, decode_cf=False)['var'].sel(lat=slice(south,north), lon=slice(west,east)).mean(dim='time')
    return mean

def diff_(x,y):
    return x-y

p1 = "/path/to/first/multi-file/dataset"
p2 = "/path/to/second/multi-file/dataset"

a = dask.delayed(calc_avg)(p1)  
b = dask.delayed(calc_avg)(p2)
total = dask.delayed(diff_)(a,b)
result = total.compute()

此处的执行时间是17s。

但是，绘制结果（result.plot（））需要超过1分钟，因此似乎在尝试绘制结果时实际上发生了计算。

这是使用DASK延迟的正确方法吗？

Answer 1

您正在打电话给 > ，它本身就是延迟函数中的Dask操作。因此，当您调用result.com pupute时，您正在执行函数calc_avg和平均值。但是，calc_avg返回DASK支持的数据。是的，17S任务将计划的延迟 calc_avg和earne的dask图将其定为计划dask.array open_mfdataset和Array OPS的DASK图。

要解决此问题，请放下延迟的包装器，然后简单地使用dask.Array Xarray工作流程：

a = calc_avg(p1)  # this is already a dask array because
                  # calc_avg calls open_mfdataset
b = calc_avg(p2)  # so is this
total = a - b     # dask understands array math, so this "just works"
result = total.compute()    # execute the scheduled job

请参阅 xarray指南与dask的并行计算指南进行简介。