Как я могу сохранить некоторое состояние объекта Python между выполнением задач Dask?

Question

У меня есть кластер работников Dask, который я хочу использовать для распараллеливания операций прогнозирования с использованием сложной модели.Файлы модели большие и требуют времени для загрузки, поэтому я использовал client.run, чтобы все работники запустили функцию инициализации для загрузки этой модели.

Как сохранить состояние переменной Python из моей функции client.runчтобы я мог обратиться к нему и использовать его в будущих операциях задачи?

Я нашел dask.distributed.get_worker и словарь worker.data и использовал его для установки произвольных значений.затем может получить доступ к map_partition функциям, но не уверен, является ли это лучшим или самым безопасным вариантом.

Если работник умирает и перезапускается, или если другие работники присоединяются к кластеру, есть ли способчтобы эти работники автоматически вызывали ту же функцию, которую я изначально передал client.run?

Answer 1

Просто используйте фьючерсы

Если ваша модель / состояние не меняется, то я, вероятно, просто использовал бы client.scatter, чтобы отправить его, и позволить Dask скопировать его по мере необходимости.Это самый простой и самый надежный подход.Если прибывает новый работник, он копирует его по мере необходимости.

Да, использование get_worker имеет смысл

Однако, если вы хотите управлять состоянием самостоятельно, тогда да, запуск функции, получение некоторыхсостояние, и присоединение его к работнику - хорошая идея:

get_worker().my_special_state = x

Я не рекомендую помещать данные в get_worker().data, потому что именно здесь Dask управляет собственной памятью.Может быть смущен, увидев там что-то постороннее, чего он не знал.Все должно быть хорошо, но вы никогда не знаете.

Рабочие плагины

Если работник умирает и перезапускается, или если другие работники присоединяются к кластеру, есть ли способПусть эти работники автоматически вызовут ту же функцию, которую я изначально передал client.run?

Да, самый простой подход здесь - использовать сценарии предварительной загрузки или рабочие плагины.См https://docs.dask.org/en/latest/setup/custom-startup.html