管理 dask localcluster 上的 worker 内存

Question

我正在尝试使用 dask 加载数据集，但是当需要计算我的数据集时，我不断遇到这样的问题：

警告 - Worker 超出了 95% 的内存预算。正在重新启动。

我只是在本地机器上工作，启动 dask 如下：

if __name__ == '__main__':
    libmarket.config.client = Client()  # use dask.distributed by default

现在，在我的错误消息中，我不断看到对“memory_limit=”关键字参数的引用。但是，我已经彻底搜索了 dask 文档，但我无法弄清楚如何在单机配置中增加血腥的工人内存限制。我有 256GB 的 RAM，在将其转换回 pandas 数据帧之前，我将删除大部分未来的列（一个 20GB 的 csv 文件），所以我知道它会适合内存。我只需要从我的代码中增加每个工作人员的内存限制（不使用 dask-worker），以便我可以处理它。

请有人帮帮我。

Answer 1

参数memory_limit可以提供给Client和LocalCluster的__init()__函数。

一般说明

只调用Client() 是先调用LocalCluster() 然后使用创建的集群(Dask: Single Machine) 调用Client 的快捷方式。当在没有LocalCluster 实例的情况下调用Client 时，可以将LocalCluster.__init()__ 的所有可能参数提供给Client 的初始化调用。因此，Client 类的 API 文档中没有记录参数 memory_limit（以及其他参数，例如 n_workers）。

但是，LocalCluster 的 API 文档中似乎没有正确记录参数 memory_limit（请参阅 Dask GitHub 问题 #4118）。

解决方案

以下是一个工作示例。我添加了更多参数，这可能对人们找到这个问题/答案很有用。

# load/import classes
from dask.distributed import Client, LocalCluster

# set up cluster and workers
cluster = LocalCluster(n_workers=4, 
                       threads_per_worker=1,
                       memory_limit='64GB')
client = Client(cluster)

# have a look at your workers
client

# do some work
## ... 

# close workers and cluster
client.close()
cluster.close()

快捷方式是

# load/import classes
from dask.distributed import Client

# set up cluster and workers
client = Client(n_workers=4, 
                threads_per_worker=1,
                memory_limit='64GB')

# have a look at your workers
client

# do some work
## ... 

# close workers and cluster
client.close()

进一步阅读

• https://distributed.dask.org/en/latest/local-cluster.html
• https://github.com/dask/dask/issues/4118