python - 关闭 Dask LocalCluster 的 "right"方法是什么？

Question

我正在尝试使用 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 问题。通常如下

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

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

这些是我正在使用的库:

• python 3.[6,7].0
• Tornado 5.1.1
• 任务 0.20.0
• 分布式 1.24.0
• matplotlib 3.0.1

感谢您的帮助!

Answer 1

扩展 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/Notebook 服务器/其他应用程序留下了一个核心。

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

• 一旦 with block 退出，cluster.close() 和 client.close() 都会被调用。第一个关闭集群、scehduler、nanny 和所有 worker，第二个断开客户端(在您的 python 解释器上创建)与集群的连接。

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

---

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

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

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.

引用资料:

• 设置本地集群:https://distributed.dask.org/en/latest/local-cluster.html
• 客户端关闭方法:https://distributed.dask.org/en/latest/api.html#distributed.Client.close

• Scheduler close 方法，根据我的理解，它是由 cluster.close() 调用的:https://distributed.dask.org/en/latest/scheduling-state.html#distributed.scheduler.Scheduler.close

• with 语句有多个变量:https://stackoverflow.com/a/1073814/4900327