numpy - 如何使用 Dask.array 有效地将大型 numpy 数组发送到集群

Question

我的本地机器上有一个大型 NumPy 数组，我想与集群上的 Dask.array 并行化

import numpy as np
x = np.random.random((1000, 1000, 1000))

但是，当我使用 dask.array 时，我发现我的调度程序开始占用大量 RAM。为什么是这样？这些数据不应该交给 worker 吗？

import dask.array as da
x = da.from_array(x, chunks=(100, 100, 100))

from dask.distributed import Client
client = Client(...)
x = x.persist()

Answer 1

每当你 persist或 compute一个 Dask 集合，数据进入调度程序，然后从那里传递给工作人员。如果您想绕过在调度程序上存储数据，那么您必须学习如何 move data与 scatter .

你有三个选择:

不要在您的客户端机器上加载数据

分散然后 block

大块然后分散

不要在您的客户端机器上加载数据

最好的方法是将加载数据作为计算的一部分，而不是在本地进行。

前

x = load_array_from_file(fn)  # load into a local numpy array
x = da.from_array(x, chunks=(100, 100, 100))  # split with dask
x = x.persist()

后

x = dask.delayed(load_array_from_file)(fn)
x = da.from_delayed(x, shape=(1000, 1000, 1000), dtype=float)
x = x.rechunk((100, 100, 100))
x = x.persist()

更多关于创建 dask 数组的信息在这里: http://dask.pydata.org/en/latest/array-creation.html

分散然后 block

您可以将 numpy 数组直接分散到工作人员

future = client.scatter(x)
x = da.from_delayed(future, shape=x.shape, dtype=x.dtype)
x = x.rechunk((100, 100, 100))
x = x.persist()

这会将您的数据直接移动到工作人员，然后从那里分 block 。这很好，因为它绕过了调度程序。但是，如果您的工作人员开始失败，您现在将面临数据丢失的风险。这仅在您处于大规模并行系统中时才重要。

这也有点低效，因为您的所有数据都集中在一个工作人员身上，而不是分散开来。您可以调用 client.rebalance或继续阅读。

block 然后分散

您可以使用本地调度程序在本地分 block 数据，然后分散到集群中。

x = da.from_array(x, chunks=(100, 100, 100))
x = x.persist(get=dask.threaded.get)  # chunk locally
futures = client.scatter(dict(x.dask))  # scatter chunks
x.dask = x  # re-attach scattered futures as task graph

留在本地

或者，您可以继续在本地使用 dask，或者使用线程调度程序，或者使用仅使用本地进程的分布式调度程序。

client = Client(processes=False)

这将停止本地进程、调度程序和工作人员之间不必要的数据复制。它们现在都在您的同一个本地进程中。

另见: How to efficiently submit tasks with large arguments in Dask distributed?对于此答案的基于任务的版本