Блог

Я пытаюсь использовать 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 рабочий завершит работу (через 20 секунд), но этого не происходит, оставаясь на целых 5 минут. Любой совет, как это исправить?

Я бы рекомендовал использовать диспетчер контекста для управления кластером — он приятный и чистый. У меня были проблемы, когда дело доходило до максимального использования оперативной памяти и зависания моего компьютера при локальной работе, но вот пример того, что я часто использую:

 # 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

 

Это приведет к завершению работы подключенного планировщика и удалению рабочих:

 C.shutdown()