dask - 如何在 Dask 中停止正在运行的任务？

Question

使用 Dask 的分布式调度程序时，我有一个正在远程工作人员上运行的任务，我想停止该任务。

我该如何阻止？我知道取消方法，但如果任务已经开始执行，这似乎不起作用。

Answer 1

如果它还没有运行

如果任务尚未开始运行，您可以通过取消关联的 future 来取消它

future = client.submit(func, *args)  # start task
future.cancel()                      # cancel task

如果您使用的是 dask 集合，那么您可以使用 client.cancel 方法

x = x.persist()   # start many tasks 
client.cancel(x)  # cancel all tasks

如果它正在运行

但是，如果您的任务已经开始在工作线程中的线程上运行，那么您无法中断该线程。不幸的是，这是 Python 的一个限制。

建立明确的停止条件

您能做的最好的事情是使用您自己的自定义逻辑在您的函数中构建某种停止标准。您可能会考虑在循环中检查共享变量。在这些文档中查找“变量”: http://dask.pydata.org/en/latest/futures.html

from dask.distributed import Client, Variable

client = Client()
stop = Varible()
stop.put(False)

def long_running_task():
    while not stop.get():
        ... do stuff

future = client.submit(long_running_task)

... wait a while

stop.put(True)