Dask计算使用的内存是预期内存的两倍。

Question

我正在努力理解和优化达斯克的内存使用。对于这个简单的例子，我创建了一个数组，我希望这个数组占用大约1GB的内存。当我意识到使用persist()的数据时，这就是我所发现的。但是有了compute()，我看到内存的数量翻了一番。

这是正确的吗？如果是，为什么？

import dask.array as da
import resource

mem0 = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
memory = lambda: (resource.getrusage(resource.RUSAGE_SELF).ru_maxrss - mem0) / 1024

data = da.random.random((1024, 1024, 128))  # ~1 GB, not yet realized
print(f"Memory used: {memory()} MB")
# Memory used: 0.49609375 MB

data.persist()
print(f"Memory used: {memory()} MB")
# Memory used: 1023.49609375 MB

# Using .compute() instead of .persist()
# Memory used: 2047.5 MB

注意:我没有在同一个脚本中运行persist和compute。实际上，我更改了代码，并再次运行，以得到公平的比较。

Answer 1

调用dask.Array.compute()时，dask完成远程工作人员上的任务，然后序列化对象，将其传回主线程，然后反序列化它。因此，很可能有一个工人数据的副本，一个副本作为字符串传递，版本在主线程上重新创建。峰值内存使用率可能超过2倍；在任何给定的时刻，远程和部分本地版本都在内存中，再加上正在传输的块的副本。

请参阅dask.distributed在管理内存上的文档。在关于数据清理的一节中：

通过从本地进程中删除集合，我们从分布式RAM中删除数据。一旦从所有客户端机器中删除指向该数据的所有期货，远程数据就会被删除。 删除本地数据的del df #经常删除远程数据 如果这是唯一的副本，那么很可能会触发集群删除数据。 但是，如果我们有多个副本或基于这个集合的其他集合，那么我们将不得不全部删除它们。 df2 = dfdf.x < 10 del df #不会删除数据，因为df2仍然跟踪期货。

在您的示例中，您从不删除对远程对象的引用，这意味着它仍然必须保存在远程集群上：

# brings the data to the local thread while keeping the remote reference
data.compute()

如果您重写了远程引用，那么在完成对compute()的调用之后，至少可以删除工作人员上的副本：

# drops the remote reference after bringing the data locally
data = data.compute()