我有一个 37 GB 的 .npy 文件，我想将其转换为 Zarr 存储，以便可以包含坐标标签。理论上我有代码可以做到这一点，但我总是内存不足。我想在中间使用 Dask 来促进分块执行此操作，但我仍然耗尽内存。
这些数据是人们股骨软骨的“厚度图”。每个地图都是一个 310x310 浮点数组，共有 47789 个这样的地图。所以数据形状是(47789,310,310)。
步骤 1：将 npy 文件加载为内存映射的 Dask 数组。
fem_dask = dask.array.from_array(np.load('/Volumes/T7/cartilagenpy20220602/femoral.npy', mmap_mode='r'),
chunks=(300, -1, -1))
步骤 2：在 Dask 数组上创建一个 xarray DataArray，并使用所需的坐标。我有几个来自“地图”维度的坐标metadata（熊猫数据框）。
fem_xr = xr.DataArray(fem_dask, dims=['map','x','y'],
coords={'patient_id': ('map', metadata['patient_id']),
'side': ('map', metadata['side'].astype(np.string_)),
'timepoint': ('map', metadata['timepoint'])
})
第三步：写信给扎尔。
fem_ds = fem_xr.to_dataset(name='femoral') # Zarr requires Dataset, not DataArray
res = fem_ds.to_zarr('/Volumes/T7/femoral.zarr',
encoding={'femoral': {'dtype': 'float32'}},
compute=False)
res.visualize()
See task graph below if desired
当我打电话时res.compute()，RAM 使用很快就会失控。其他 python 进程（我认为是 Dask 工作进程）似乎处于不活动状态：
But a bit later, they are active -- see that one of those Python processes now has 20 gb RAM and another has 36 gb:
我们还可以从 Dask 仪表板确认这一点：
最终所有的工人都被杀了，任务也出错了。我怎样才能以有效的方式正确使用 Dask、xarray 和 Zarr，而不耗尽 RAM（或熔化笔记本电脑）？
如果 dask 工作人员可以共享线程，那么您的代码应该可以正常工作。如果您没有显式初始化 dask 集群，dask.Array 将使用默认参数创建一个使用进程的集群。这会导致您所看到的行为。要解决此问题，请使用线程显式创建集群：
# use threads, not processes
cluster = dask.distributed.LocalCluster(processes=False)
client = dask.distributed.Client(cluster)
arr = np.load('myarr.npy', mmap_mode='r')
da = dda.from_array(arr).rechunk(chunks=(100, 310, 310))
da.to_zarr('myarr.zarr', mode='w')
如果您使用的集群无法共享线程，例如 JobQueue、KubernetesCluster 等，则可以使用以下命令来读取 npy 文件，假设它位于网络文件系统上或以某种方式可供所有工作人员使用。
这是一个工作流程，从内存映射创建一个空数组，然后使用映射读取作业dask.array.map_blocks。关键是使用block_info可选关键字，它提供有关数组中块的位置的信息，我们可以使用 dask 工作程序来切片新的 mmap 数组对象：
def load_npy_chunk(da, fp, block_info=None, mmap_mode='r'):
"""Load a slice of the .npy array, making use of the block_info kwarg"""
np_mmap = np.load(fp, mmap_mode=mmap_mode)
array_location = block_info[0]['array-location']
dim_slicer = tuple(list(map(lambda x: slice(*x), array_location)))
return np_mmap[dim_slicer]
def dask_read_npy(fp, chunks=None, mmap_mode='r'):
"""Read metadata by opening the mmap, then send the read job to workers"""
np_mmap = np.load(fp, mmap_mode=mmap_mode)
da = dda.empty_like(np_mmap, chunks=chunks)
return da.map_blocks(load_npy_chunk, fp=fp, mmap_mode=mmap_mode, meta=da)
这对我来说适用于相同大小的演示（您可以在最后添加 xarray.DataArray 创建/格式化步骤，但 dask 操作工作正常，并且工作内存对我来说保持在 1GB 以下）：
import numpy as np, dask.array as dda, xarray as xr, pandas as pd, dask.distributed
### insert/import above functions here
# save a large numpy array
np.save('myarr.npy', np.empty(shape=(47789, 310, 310), dtype=np.float32))
cluster = dask.distributed.LocalCluster()
client = dask.distributed.Client(cluster)
da = dask_read_npy('myarr.npy', chunks=(300, -1, -1), mmap_mode='r')
da.to_zarr('myarr.zarr', mode='w')