在工作完成后终止DASK工人

问题描述

我正在尝试使用集群上的Dask.我有兴趣在所有工作完成后立即终止所有工作。 我试着用退休工人的方法来做到这一点，但这似乎并没有杀死工人。 这里有一个例子。

import time
import os
from dask.distributed import Client

def long_func(x):
    time.sleep(2)
    return 1

if __name__ == '__main__':
    C = Client(scheduler_file='sched.json')
    res = []
    for _ in range(10):
        res.append(C.submit(long_func, _))
    for r in res:
        r.result()
    workers = list(C.scheduler_info()['workers'])
    # C.run(lambda: os._exit(0), workers=workers)
    C.retire_workers(workers=workers, close_workers=True)

使用以下命令启动调度程序和工作程序：

dask-scheduler --scheduler-file sched.json 
dask-worker --scheduler-file sched.json --nthreads=1 --lifetime='5minutes'

希望在执行上面的python代码之后，Worker将终止(20秒后)，但它并没有，整个5分钟都停留在那里。有什么建议可以解决这个问题吗？

---

解决方案

这将关闭连接的计划程序并使工作进程停用：

C.shutdown()