Распараллелить обработку небольших данных с использованием большой модели с фьючерсами Dask (рабочий контекст) - PullRequest
Новая
       11

Распараллелить обработку небольших данных с использованием большой модели с фьючерсами Dask (рабочий контекст)

0 голосов
/ 11 декабря 2018

Я попытался распараллелить свою обработку, используя Dask: 

engine = init_my_engine_with_large_NN_models(models) # takes seconds
for image in images:  # there are thousands of images
  dask.distributed.Client().submit(engine.process, image)

Каждый вызов передает «механизм», который должен быть воссоздан (сериализован / десериализован) в каждом работнике, и это занимает в несколько раз больше времени, чем обработка изображения,В итоге мой код потребляет гораздо больше энергии и займет больше времени, чем однопоточная обработка по умолчанию в python.

Метод Client.map () тоже не помогает.Я протестировал его, он сериализует метод с 'self = engine' для каждого изображения в итерируемом 'images'. 

dask.distributed.Client().map(engine.process, images)

Я хотел бы иметь возможность сериализовать и настроить мой 'engine' для всех работников Dask простоодин раз и есть какой-то метод, с помощью которого я могу получить свой движок во время обработки на работнике, что-то вроде: 

engine = init_my_engine_with_large_NN_models(models)
client = dask.distributed.Client()
client.send_to_workers("engine", engine)
# engine is deserialized on each worker under the name "engine"
for image in images:  # there are thousands of images
  client.submit(
      lambda im: get_client().get_from_worker("engine").process(im), image)

Мой образ мышления неправильный?У вас есть другой рабочий процесс / методология, как реализовать эффективное распараллеливание такой обычной проблемы?

РЕДАКТИРОВАНИЕ:

Я проверил concurrent.futures и там ProcessPoolExecutor (initializer = my_initializer) добавлен в последних версиях Python для этой цели.

Теперь я понимаю, что вы можете сделать подобное в Dask , используя Client.run () метод: 

client.run(lambda engine: dask.distributed.get_worker().
           __dict__.update({"juraj_engine" : engine}), engine)

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

Пожалуйста, разместите лучшие (элегантные) решения.

...