Блог

У меня есть следующий сценарий, который мне нужно решить с помощью планировщика Dask и рабочих:

• Программа Dask имеет N функций, вызываемых в цикле (N определяется пользователем)
• Каждая функция запускается delayed(func)(args) для параллельного выполнения.
• Когда запускается каждая функция из предыдущей точки, она запускает W рабочих. Вот как я призываю работников:

   futures = client.map(worker_func, worker_args)     
  worker_responses = client.gather(futures)
   

Это означает, что мне нужны N * W работников, чтобы запускать все параллельно. Проблема в том, что это не оптимально, так как это слишком большое распределение ресурсов, я запускаю его в облаке, и это дорого. Кроме того, N определяется пользователем, поэтому я заранее не знаю, какими возможностями обработки мне нужно обладать.

Есть ли способ поставить работников в очередь таким образом, чтобы, если я определю, что в Dask есть X работников, когда один работник заканчивается, начинается следующий?

Сначала определите необходимое вам количество работников, относитесь к ним как к эфемерным, но статичным на протяжении всего процесса обработки
Вы можете создавать их динамически (при запуске или позже), но, вероятно, захотите, чтобы все они были готовы в самом начале обработки

С вашей точки зрения, клиент является исполнителем (поэтому, когда вы ссылаетесь на работников и работаете параллельно, вы, вероятно, имеете в виду одно и то же

Этот класс напоминает исполнителей, concurrent.futures но также допускает Future объекты в submit/map вызовах. Когда создается экземпляр клиента, он берет на себя все dask.compute и dask.persist вызывает по умолчанию.

Как только ваши работники будут доступны, Dask распределит работу, предоставленную им через планировщик

Вы должны выполнять любые задачи, которые зависят друг от друга, передавая результат dask.delayed() с помощью предшествующей функции result (которая является будущим, а еще не результатом)
. Это будущее в качестве аргументов позволит Dask построить график задач вашей работы

Пример использования https://examples.dask.org/delayed.html
Ссылка на будущее https://docs.dask.org/en/latest/futures.html#distributed.Будущее

Зависимое будущее с даском.отложено

Вот полный пример из отложенных документов (на самом деле объединяет несколько последовательных примеров с одним и тем же результатом)

 import dask
from dask.distributed import Client

client = Client(...)  # connect to distributed cluster

def inc(x):
    return x   1

def double(x):
    return x * 2

def add(x, y):
    return x   y

data = [1, 2, 3, 4, 5]

output = []
for x in data:
    a = dask.delayed(inc)(x)
    b = dask.delayed(double)(x)
    c = dask.delayed(add)(a, b)    # depends on a and b
    output.append(c)

total = dask.delayed(sum)(output)  # depends on everything
total.compute()  # 45
 

Вы можете позвонить total.visualize() , чтобы просмотреть график задач

(изображение из документов Dask с задержкой)

Коллекции фьючерсов

Если вы уже используете .map(..) для сопоставления пар функций и аргументов, вы можете продолжать создавать фьючерсы, а затем .gather(..) их все сразу, даже если они находятся в коллекции (что удобно для вас здесь).

.gather() «Эд results » будет располагаться в том же порядке, в каком они были даны (список списков).

 [[fn1(args11), fn1(args12)], [fn2(args21)], [fn3(args31), fn3(args32), fn3(args33)]]
 

https://distributed.dask.org/en/latest/api.html#distributed.Клиент.соберите

 import dask
from dask.distributed import Client

client = Client(...)  # connect to distributed cluster

collection_of_futures = []

for worker_func, worker_args in iterable_of_pairs_of_fn_args:
    futures = client.map(worker_func, worker_args)
    collection_of_futures.append(futures)

results = client.gather(collection_of_futures)
 

Примечания

• worker_args должно быть какое-то повторяющееся отображение worker_func , которое может быть источником ошибок
• .gather() ing будет блокироваться до тех пор, пока все фьючерсы не будут завершены или не повысятся

.as_completed()

Если вам нужны результаты как можно быстрее, вы можете использовать .as_completed(..) , но обратите внимание, что результаты будут в недетерминированном порядке, поэтому я не думаю, что это имеет смысл для вашего случая .. если вы обнаружите, что это так, вам понадобятся некоторые дополнительные гарантии

• включите в результат информацию о том, что делать с результатом
• держите ссылку на каждый из них и проверяйте их
• объединяйте группы только там, где это не имеет значения (т. е. все фьючерсы имеют одну и ту же цель)

также обратите внимание, что данные фьючерсы завершены, но все еще являются будущим, поэтому вам все равно нужно позвонить .result() или .gather() им

https://distributed.dask.org/en/latest/api.html#distributed.as_completed