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

【问题标题】：Dask and numpy - slow conversion between numpy array and dask arrayDask 和 numpy - numpy 数组和 dask 数组之间的缓慢转换
【发布时间】：2020-02-20 00:01:41
【问题描述】：

我需要从一个大的 numpy 数组中保存一个 dask 数组。下面有一个显示该过程的最小工作示例。请注意，a 是使用 numpy.random 创建的，仅用于此 mwe，不幸的是我无法使用 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，但是在运行 dask 部分时，使用的内存开始上升，直到它几乎填满可用的 ram，即超过 300GB。然后它做了一些计算，一段时间后（比如 10 分钟）我得到了一个内存错误。我需要由 dask 保存的数组，因为我有另一个使用 dask 数组的管道（不能直接连接到该管道）并从内存中读取 dask 数组，需要info 文件（如果有任何其他转储方法数组并创建info 文件，我打开试试看）。

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

【问题讨论】：

不是用 numpy 创建数组，你可以用 dask 创建吗？ > arr = da.random.randint(0,2,size=4000000*8000).reshape((4000000,8000))
不，很遗憾，我无法使用 dask 创建数组。 @quasiben
你的数据读得怎么样？
数据来自torch。但这超出了这个问题的范围。我也可以接受不能以这种方式使用 dask，我只是在寻找答案。谢谢@quasiben
examples.dask.org/machine-learning/torch-prediction.html 中的任何技巧对您有用吗？

标签： python numpy dask dask-distributed

【解决方案1】：

如果您在单台机器上，那么我建议使用标准线程调度程序而不是dask.distributed.Client。您将通过这种方式将所有数据保存在同一进程中，并且无需复制大型 Numpy 数组。

【讨论】：

我会试试的。同时让我问一下为什么 dask 开发人员建议即使在单台机器上也始终使用分布式调度程序@MRocklin
我个人不建议你总是使用分布式调度程序，只是它通常是一个非常好的选择。没有任何解决方案在所有情况下都是最佳选择。

【解决方案2】：

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

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

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

【讨论】：

（我不确定这些块，你想要那么多原创作品还是输出作品）
感谢@mdurant，但是正如我在主要问题正文中所评论的那样，我无法直接使用 dask 创建数组。该代码只是运行它的一个示例。数据来自另一个管道。