从未来创建惰性xarray对象

Question

我有一个dask.delayed函数，它接受一个xarray.Dataarray作为参数并返回一个参数

我正在创建一些延迟任务，并使用dask.distributed将它们传递给client.compute。每次调用compute都会返回一个distributed.client.Future，表示将返回的数据数组

我的问题是:

有没有一种方法可以在将来再次构建一个“惰性”数据数组，而无需从工作者加载实际数据？我的意图是基于第一次计算的输出构建第二个任务图

我已经看到了client.gather，但这似乎将所有数据拉回到了客户端，这不是我想要的

下面是一个小例子：

import dask
from distributed import Client
import xarray as xr

# load example data
x = xr.tutorial.open_dataset("air_temperature")

# use first timestep
x_t0 = x.isel(time=0)

# delayed 'processing' function
@dask.delayed
def fun(x):
    return x*2

# init client
client = Client()

# compute on worker
future = client.compute(fun(x_t0))

# when done
print(future)
# <Future: finished, type: xarray.Dataset, key: fun-96cd56f4-4ed3-4eac-ade9-fe3f17e4b8c6>

## now how to get back to lazy xarray from future?

Answer 1

我不知道你到底想达到什么目的。也许有更好的方法可以做到这一点，而不是从未来创建新阵列。也就是说，这将从您的数据创建一个新的数据数组： 您不必调用compute来保持它的惰性

（如果需要dask阵列而不是xarray阵列，请删除xr.DataArray）

import dask
from distributed import Client
import xarray as xr

# load example data
x = xr.tutorial.open_dataset("air_temperature")

# use first timestep
x_t0 = x.isel(time=0)

# delayed 'processing' function
@dask.delayed
def fun(x):
    return x*2

# init client
client = Client()

# Create lazy xarray object from future:
import dask.array as da

new_ds = xr.DataArray(da.from_delayed(client.persist(fun(x_t0)), shape=x_t0.air.shape, meta='f8'), coords=x.coords)

编辑：已添加客户端。保存以将数据保留在客户端上

输出：