什么是关闭Dask LocalCluster的"正确"方法？

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.application - ERROR - Exception in Future <Future cancelled> after timeout

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

这些是我正在使用的库:

• python 3. [6,7] .0
• 龙卷风5.1.1
• dask 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.

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

• 每个 worker 初始化为 2GB 的 RAM。拥有一个临时集群允许您为不同的工作负载使用不同数量的 RAM 来启动工作程序。

• 一旦with块退出，cluster.close()和 都会client.close()被调用。第一个关闭集群、sehduler、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.

参考：

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

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

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

Answer 2

根据我们的经验 - 最好的方法是使用上下文管理器，例如：

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
    )
    with Client(cluster) as client:
        x = np.linspace(0, 1, 100)
        y = x * x
        plt.plot(x, y)
        print('Computation complete! Stopping workers...')

    print('Execution complete!')