How to use Dask Distributed Futures

Question

I have a future that is a python set that I broadcasted (LocalCluster):

In [0]: [set_future] = client.scatter([_set], broadcast=True)

In [1]: set_future

Out[1]: Future: set status: finished, type: builtins.set, key: set-529f704c52fef330450e5d68302fbeac

Now I simply want to have that data available in my map_partitons op:

In [2]: def mapper(pdf, _set):
          assert type(_set)==set
          return pdf

        ddf.map_partitions(mapper, set_future)

Out[2]: AssertionError()

However, in the mapper the type is distributed.client.Future and not set. The future doesn't seem to be recovered from the cluster. What am I doing wrong?

eggie5 eggie5 · Accepted Answer · 2020-05-04T19:14:02

If you don't provide meta to map_partitions then Dask will try to infer it w/ dummy data and will actually evaluation your mapping function. However, in this context, the futures don't resolve from the cluster which will cause your function to error.

In summary, if you are using futures, you must provide meta.

How to use Dask Distributed Futures

1 Answers