时间戳

首页 > 解决方案 > 关闭 Dask LocalCluster 的“正确”方法是什么？

python - 关闭 Dask LocalCluster 的“正确”方法是什么？

问题描述

我正在尝试使用 LocalCluster 在我的笔记本电脑上使用 dask-distributed，但我仍然没有找到一种方法让我的应用程序关闭而不引发一些警告或触发一些奇怪的迭代 matplotlib（我正在使用 tkAgg 后端）。

例如，如果我按此顺序关闭客户端和集群，则 tk 无法以适当的方式从内存中删除图像，我收到以下错误：

Traceback (most recent call last):
  File "/opt/Python-3.6.0/lib/python3.6/tkinter/__init__.py", line 3501, in __del__
    self.tk.call('image', 'delete', self.name)
RuntimeError: main thread is not in main loop

例如，以下代码会生成此错误：

from time import sleep
import numpy as np
import matplotlib.pyplot as plt
from dask.distributed import Client, LocalCluster

if __name__ == '__main__':
    cluster = LocalCluster(
        n_workers=2,
        processes=True,
        threads_per_worker=1
    )
    client = Client(cluster)

    x = np.linspace(0, 1, 100)
    y = x * x
    plt.plot(x, y)

    print('Computation complete! Stopping workers...')
    client.close()
    sleep(1)
    cluster.close()

    print('Execution complete!')

该sleep(1)行使问题更容易出现，因为它不会在每次执行时发生。

我试图停止执行的任何其他组合（避免关闭客户端、避免关闭集群、避免同时关闭两者）都会产生龙卷风问题。通常如下

tornado.application - ERROR - Exception in Future <Future cancelled> after timeout

停止本地集群和客户端的正确组合是什么？我错过了什么吗？

这些是我正在使用的库：

谢谢您的帮助！

标签： pythondaskdask-distributed

解决方案

扩展skibee的答案，这是我使用的一种模式。它设置了一个临时的 LocalCluster，然后将其关闭。当您的代码的不同部分必须以不同的方式并行化时非常有用（例如，一个需要线程，另一个需要进程）。

from dask.distributed import Client, LocalCluster
import multiprocessing as mp

with LocalCluster(n_workers=int(0.9 * mp.cpu_count()),
    processes=True,
    threads_per_worker=1,
    memory_limit='2GB',
    ip='tcp://localhost:9895',
) as cluster, Client(cluster) as client:
    # Do something using 'client'

上面发生了什么：

  • 一个集群正在您的本地机器（即运行 Python 解释器的机器）上启动。此集群的调度程序正在侦听端口 9895。

  • 创建了集群并启动了许多工作人员。每个工人都是一个进程，因为我指定了processes=True.

  • 启动的工人数量是 CPU 核心数量的 90%，向下取整。因此，一台 8 核机器将产生 7 个工作进程。这至少为 SSH/笔记本服务器/其他应用程序留出了一个内核。

  • 每个工作人员都使用 2GB 的 RAM 进行初始化。拥有一个临时集群允许您为不同的工作负载启动具有不同 RAM 量的工作人员。

  • 一旦with块退出，两者cluster.close()都client.close()被调用。第一个关闭集群、schehduler、nanny 和所有工作人员，第二个将客户端（在您的 python 解释器上创建）与集群断开连接。

当工作集正在处理时，您可以通过检查来检查集群是否处于活动状态lsof -i :9895。如果没有输出，则集群已关闭。

示例用例：假设您想使用预训练的 ML 模型来预测 1,000,000 个示例。

该模型经过优化/矢量化，因此它可以非常快速地预测 10K 示例，但 1M 很慢。在这种情况下，可行的设置是从磁盘加载模型的多个副本，然后使用它们来预测 1M 示例的块。

Dask 允许您轻松完成此操作并实现良好的加速：

def load_and_predict(input_data_chunk):
    model_path = '...' # On your disk, so accessible by all processes.
    model = some_library.load_model(model_path)
    labels, scores = model.predict(input_data_chunk, ...)
    return np.array([labels, scores])

# (not shown) Load `input_data`, a list of your 1M examples.

import dask.array as DaskArray

da_input_data = DaskArray.from_array(input_data, chunks=(10_000,))

prediction_results = None
with LocalCluster(n_workers=int(0.9 * mp.cpu_count()),
    processes=True,
    threads_per_worker=1,
    memory_limit='2GB',
    ip='tcp://localhost:9895',
) as cluster, Client(cluster) as client:
    prediction_results = da_input_data.map_blocks(load_and_predict).compute()

# Combine prediction_results, which will be a list of Numpy arrays, 
# each with labels, scores for 10,000 examples.

参考：

推荐阅读