Стоимость доступа к разбросанным данным в кластере dask

Question

Я использую dask для распараллеливания некоторой обработки, что очень радует.

У меня есть случай, когда для расчета на стороне клиента требуются некоторые данные для поиска, которые достаточно тяжело генерировать, поэтому рассредоточьте эти данные по клиентам:

[future_dict] = client.scatter([large_dict], broadcast=True)

Тогда вычисление что-то вроде

def worker(i):
    key = do_some_work()
    data = future_dict.result()[key] 
    res = do_some_more_work( data )
    return (i, res )

f = client.map( worker, range(200))
res = client.gather( f )

Это работает, но поиск future_dict.result()[key] довольно медленный. Время, которое требуется для поиска в рабочем файле, похоже на удаление выбранной версии large_dict, поэтому я предполагаю, что мой словарь десериализован для каждого рабочего.

Могу ли я сделать что-нибудь для доступа разбросать данные быстрее? Например, если моя гипотеза о том, что десериализованные данные у каждого работника верны, могу ли я сделать что-то, чтобы десериализация происходила только один раз для каждого клиента?

Answer 1

То, что вы делаете, должно быть в порядке, но если вы хотите сделать это быстрее, вы можете передать в будущем явный аргумент.

def func(i, my_dict=None):
    key = do_some_work()
    data = my_dict[key] 
    res = do_some_more_work( data )
    return (i, res )

f = client.map( func, range(200), my_dict=future_data)
res = client.gather( f )