Dask和numpy-numpy数组和dask数组之间的缓慢转换

Question

我需要从一个大的numpy数组中保存一个dask数组。下面是显示该过程的最低限度的工作示例。请注意，a仅针对此mwe用numpy.random创建，不幸的是，我无法使用dask创建数组。

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

a = numpy.random.randint(0,2,size=4000000*8000).reshape((4000000,8000))
# here the conversion and saving
client = Client(n_workers=90, threads_per_worker=20, processes=True)
dask_array = da.from_array( a, chunks = 100000)
da.to_npy_stack(‘my-folder/’, dask_array)
client.close()

我面临的问题是，内存中的a大约需要100GB，但是在运行轻快部分时，所用的内存就会开始增加，直到将可用内存几乎填满为止，即超过300GB。然后进行一些计算，经过一段时间（例如10分钟），我收到了内存错误。我需要由dask保存的数组，因为我还有另一个使用dask数组的管道（无法直接连接到该管道），并且需要从内存中读取dask数组，因此需要info文件（如果有其他方法可以转储）数组并创建info文件，我可以尝试打开它。

关于如何加快和解决此任务的任何建议？

Answer 1

在主流程中创建所有数据，然后将其上传到辅助流程是一个坏主意！您应始终努力直接在工作线程中加载/创建数据，这将a）避免重复工作和复制数据，b）使数据保持惰性，仅在需要时将其具体化到内存中。

在这种情况下，可能看起来像

arr = da.random.randint(0, 2, size=4000000*8000, chunks=100000).reshape((4000000,8000))

Answer 2

如果您在一台机器上，那么我建议使用标准的线程调度程序，而不要使用dask.distributed.Client。您将以这种方式将所有数据保留在相同的过程中，从而无需制作大型Numpy阵列的副本。