Unable to use published datasets in a different client

[nikhilrajrs]

I get the error "Inputs contain futures that were created by another client" when I try to join a published dataset within a client different from the one that originally published it. The whole flow can be summarised as follows:

Client 1
df = dd.read_csv(...)
client.persist(df)
client.publish_dataset(ds_name=df)

Client 2
df = client.get_dataset("ds_name")
df2 = dd.read_csv(...)
df2.join(df)
client.persist(df2)

Is this expected behavior?

For reference, this check was added in the commit c02ea63#diff-96a27223dc91b5c9ea3d03684d79ad3f%5D which is part of the pull request #2227

[VMois]

Are your code for publishing dataset should looks like this:

df = dd.read_csv(...)  
df = client.persist(df) 
client.publish_dataset(my_dataset=df)  

Because client.persist() returns futures and you need to reference those futures for publishing dataset, but in your flow example you just call client.persist() without assigning it to any new variable. Please, check docs examples (https://distributed.readthedocs.io/en/latest/api.html#distributed.Client.publish_dataset) and add your actual code.

[nikhilrajrs]

Sorry for being not precise enough.
The code I use is indeed:

names = dd.read_csv('/shared-store-path/names.csv')
names = names.set_index("ID")
names = client1.persist(names)
client1.publish_dataset(names=names)

I have attached a full example that can be used to reproduce the behavior. Shown below is the output from the notebook I used to test -

from dask.distributed import Client
from dask.distributed import wait
import dask.dataframe as dd

client1 = Client("scheduler-address:8786")
client2 = Client("scheduler-address:8786")

names = dd.read_csv('/shared-store-path/names.csv')
names = names.set_index("ID")
names = client1.persist(names)
wait(names)

DoneAndNotDoneFutures(done={<Future: status: finished, type: DataFrame, key: ('sort_index-6f9583110f3b77c24727c1e970735470', 0)>}, not_done=set())

client1.publish_dataset(names=names)

roles = dd.read_csv('/shared-store-path/roles.csv')
roles = roles.set_index("ID")

names_dataset = client2.get_dataset("names")

roles = roles.join(names_dataset)

roles.head()

	Role	Name
ID
1	Developer	John
2	Developer	Doe
3	Developer	Johnny
4	Developer	Doey
5	Tester	Hello

client2.persist(roles)

---------------------------------------------------------------------------

ValueError                                Traceback (most recent call last)

<ipython-input-20-7cb62cd7bd89> in <module>()
----> 1 client2.persist(roles)


/misc/anaconda3-amp/lib/python3.6/site-packages/distributed/client.py in persist(self, collections, optimize_graph, workers, allow_other_workers, resources, retries, priority, fifo_timeout, actors, **kwargs)
   2561                                          user_priority=priority,
   2562                                          fifo_timeout=fifo_timeout,
-> 2563                                          actors=actors)
   2564 
   2565         postpersists = [c.__dask_postpersist__() for c in collections]


/misc/anaconda3-amp/lib/python3.6/site-packages/distributed/client.py in _graph_to_futures(self, dsk, keys, restrictions, loose_restrictions, priority, user_priority, resources, retries, fifo_timeout, actors)
   2210                     msg = ("Inputs contain futures that were created by "
   2211                            "another client.")
-> 2212                     raise ValueError(msg)
   2213 
   2214             if restrictions:


ValueError: Inputs contain futures that were created by another client.

testfiles.zip

[VMois]

Okay. Thanks for full example. I will try to run it tomorrow and see what will happen.

[VMois]

Actually, I have reproduced your error, but not using your code. It enough to publish a dataset with one client, then using second client (different process) get a dataset, modify something and if you will try to make client.persist() it will throw above error. I think it's a general error. My small example:

client = Client('localhost:8786')
client.restart()

df = dd.read_parquet('test_data/data.parquet')
df = client.persist(df)
client.publish_dataset(test=df)

Now, you can open a new terminal:

client = Client(...)
df = client.get_dataset('test')
new_df = df['some'] > 0.5
new_df = client.persist(new_df)
# error here...

Do you have any ideas @mrocklin ?

Python: 3.6.6
Dask: 0.20.1
Dask distributed: 1.24.1

[VMois]

Any updates/suggestions about this problem @mrocklin @martindurant ?

[martindurant]

(I am at a conference and unlikely to be able to look into this right now)

[itozapata]

did anything ever happen with this ticket. i seem to have run into the same situation and dont see any information anywhere else about it.

My situation is i am persisting a dataset the first time i load it and someone else is picking it up and adding more to the graph something akin to

df = client.get_dataset('test')
new_df = df['some'] > 0.5
new_df = client.compute(new_df)

and it is giving me this error

[martindurant]

The original posting above used versions that are now somewhat out of date. Would you mind trying with the latest versions? I just ran a very similar test and didn't see any error.

[itozapata]

sounds good after continuing to look i think i am getting something just slightly different. i will try and package up an example. I was able to get it working by not creating my own client and instead using get_client on my computer to make sure that the same client was being used with the client.compute call, but the behavior that i was experiencing didnt seem to make sense to me.

I will attempt to get something small to illustrate the situation shortly

[tshatrov]

We're also having troubles working with this check. There's a bunch of datasets published by a loader script, and another application is launching tasks that use .loc[...].compute() to retrieve a part of published dataset as a pandas dataframe and then perform some calculations with that dataframe. The way I see it, if the published dataset contains futures, and dask-worker retrieves it, then future.client is always a different object than the client of the worker, so future.client is self check will always fail. So how are we supposed to use published datasets from within workers?

[mrocklin]

future.client should become the client on the local worker.

[tshatrov]

That's not the case, the futures from the retrieved dataframe have a client with a different id (which is different for every session)

In [1]: from distributed import Client

In [2]: cl = Client("*redacted*")

In [3]: cl.id
Out[3]: 'Client-99fd4b86-8c42-11e9-b54a-48df3757a8a0'

In [4]: df = cl.get_dataset('test')

In [5]: list(df.dask.values())[0].client.id
Out[5]: 'Client-aaa7eacc-8c42-11e9-b54a-48df3757a8a0'

In [7]: dask.__version__
Out[7]: '1.2.2'

In [9]: distributed.__version__
Out[9]: '1.28.1'

[TomAugspurger]

Can anyone provide a reproducible example? This doesn't do it, I suspect because we aren't getting references to the client in the task graph

from distributed import LocalCluster, Client
import dask.array as da


def main():
    cluster = LocalCluster(dashboard_address=None)
    data = da.ones(3)

    with Client(cluster) as a:
        a.publish_dataset(data=data)
        with Client(cluster, set_as_default=False) as b:
            print(b.get_dataset('data').compute())


if __name__ == '__main__':
    main()

[TomAugspurger]

Oh @tshatrov, can you repeat your test in #2336 (comment) after calling optimize on the graph? That's when things would be replaced:

distributed/distributed/client.py

Line 2519 in 587be8d

def _optimize_insert_futures(self, dsk, keys):

[tshatrov]

STR:

1. Launch dask-scheduler and dask-worker pointing at dask-scheduler.
2. Launch ipython and do the following

import dask
from dask.distributed import Client

uri = <dask-scheduler's uri>
cl = Client(uri)
ts = dask.datasets.timeseries()
ts = ts.persist()
cl.publish_dataset(timeseries=ts)

def get_ts():
    from dask.distributed import worker_client
    with worker_client() as client:
       return client.get_dataset('timeseries').compute()

cl.submit(get_ts).result()

As a result I get ValueError: Inputs contain futures that were created by another client.

I can't reproduce this on LocalCluster.

I have not figured out how to use optimize_insert_futures to make this not fail. As far as I see it doesn't change the dataframe.