带有上下文管理器的 dask 集群答案

【问题标题】：dask clusters with context manager带有上下文管理器的 dask 集群
【发布时间】：2021-03-12 07:04:55
【问题描述】：

考虑这样一个简单的工作流程：

from dask.distributed import Client
import time

with Client() as client:
    futs = client.map(time.sleep, list(range(10)))

由于上下文管理器将关闭，上述代码将提交并几乎立即取消期货。在使用client.gather 完成任务之前，可以保持上下文管理器处于打开状态，但这会阻止当前进程中的进一步执行。

我有兴趣在同一个进程中向多个集群（例如本地和分布式）提交任务，最好不要阻塞当前进程。明确定义不同的客户端和集群很简单，但是否也可以使用上下文管理器（每个唯一的客户端/集群一个）？

这听起来有点像反模式，但也许只有在计算所有期货运行后才能关闭集群。我试过fire_and_forget，也试过传递shutdown_on_close=False，但似乎没有实现。

【问题讨论】：

我想我们可以同意这是一个反模式 - 为什么你要专门使用 with 上下文？
问题是我在 SLURM 集群上运行，如果我没有正确关闭集群，例如因为我的主进程由于内存不足错误而被杀死，我有jobs（启动 dask 工作人员）留在队列中，需要手动清除......这是一个小麻烦，所以我想也许可以修复使用上下文管理器，但我知道它现在是一种反模式。
也许使用weakref.finalize 或atexit？另外，我认为您可以将回调附加到您的未来，以便在完成后发生一些事情。
感谢您的指点，我将探索这些选项。

标签： dask dask-distributed fire-and-forget

【解决方案1】：

对于某些 Dask 集群/调度程序类型，例如 dask-cloudprovider ECSCluster，上述使用 with 块和 shutdown_on_close=False 的方法可以正常工作。

ECSCluster 和 SLURMCluster 都派生自 SpecCluster。但是，ECSCluster 通过此调用将其**kwargs（包括shutdown_on_close）传递给SpecCluster 构造函数：

super().__init__(**kwargs)

（见ECSCluster代码here）

SLURMCluster 没有：它调用JobQueueCluster 构造函数，而后者又用其参数的子集实例化SpecCluster：

super().__init__(
    scheduler=scheduler,
    worker=worker,
    loop=loop,
    security=security,
    silence_logs=silence_logs,
    asynchronous=asynchronous,
    name=name,
)

查看JobQueueCluster 代码here

因此SLURMCluster/JobQueueCluster 忽略shutdown_on_close（和其他可选参数）。您的用例似乎需要更新到 JobQueueCluster。

【讨论】：