Cost of accessing scattered data in a dask cluster

user_11246330 0
Viewed 109

I use dask to parallelize some processing, which is quite a joy.

I have an case, where the calculation on the client side requires some lookup data that is quite heavy to generate, so scatter these data to the clients:

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

The calculation is then something like

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 )

This works, but the lookup future_dict.result()[key] is quite slow. The time it takes to do the lookup in the worker is similar to unpickl'ing a pickled version of large_dict, so I guess my dictionary de-serialized in each worker.

Can I do anything to make access to scattered data faster? Eg if my hypothesis of the data being de-serialized in each worker is correct, can I the do something to make the de-serialization happen once in each client only?

dask-distributed dask python
1 Answers

What you're doing should be ok, but if you wanted to make it faster you could pass in the future an explicit argument. 

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 )
user_6166160
Related