Можно ли в Dask выбрать воркеров для конкретных c задач?

Question

У меня есть процесс, который я запускаю в своем кластере Kubernetes с Dask, который состоит из двух этапов сокращения карты, но обе карты по узлам загружают потенциально многочисленные большие файлы каждому рабочему. Чтобы избежать того, чтобы две разные машины обрабатывали одно и то же подмножество файлов на двух разных этапах карты, можно ли детерминированно выбрать, какие рабочие будут получать аргументы для одних и тех же заданий? Концептуально то, что я хочу, могло бы быть примерно таким:

workers : List = client.get_workers();
#                       ^^^^^^^^^^^
filenames : List[str] = get_filenames(); # input data to process

# map each file to a specific worker
file_to_worker = { filename : workers[hash(filename) % len(workers)] for filename in filenames }

# submit each file, specifying which worker should be assigned the task
futures = [client.submit(my_func, filename, worker=file_to_worker[filename]) for filename in filenames]
#                                           ^^^^^^ 

Что-то вроде этого позволило бы мне направлять разные этапы вычислений для одних и тех же файлов на одни и те же узлы, устраняя необходимость в повторном кэшировании файлов. .

Answer 1

да, вы можете отправлять функции в указанные c рабочие:

c .run (fun c, worker = [WorkerA, WorkerB, WorkerC])

Вы также можете прикрепить ресурсы метаданных к работникам и отправить их с этим определением вместо указанных c имен хостов:

data = [client.submit(load, fn) for fn in filenames]
processed = [client.submit(process, d, resources={'GPU': 1}) for d in data]
final = client.submit(aggregate, processed, resources={'MEMORY': 70e9})

Информацию о настройке см. В ресурсной документации

Answer 2

Вы можете указать это, используя сходство узла