如何解决在工作完成后终止 dask 工人
我正在尝试在集群上使用 dask,并且我有兴趣在所有工作完成后立即终止所有工作人员。 我试图用 retire_workers 方法来做到这一点,但这似乎并没有杀死工人。 这是一个例子。
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()
,
我建议使用上下文管理器来管理集群 - 它既漂亮又干净。我在本地工作时遇到了 RAM 内存最大化和计算机停顿的问题,但这里有一个我经常使用的示例:
# start our Dask cluster
from dask.distributed import Client,LocalCluster
if __name__ == '__main__':
cluster = LocalCluster()
with Client(cluster) as client:
print("scheduler host: ",client.scheduler.address)
# do some stuff
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。