Modifications to published, persisted datasets don't seem to visible in another client.

[funnydevnull]

I may well be misunderstanding something or doing something wrong but I want to have multiple clients work on a shared dataset so this is the approach I take:

client1 = Client('dask-scheduler:8786')
client2 = Client('dask-scheduler:8786')
cx=da.from_array(np.random.normal(0,1, (1000,100)), (10,100))
cxp=client1.persist(cx)
client1.publish_dataset(myarr=cxp)
x2=client2.get_dataset('myarr')

# check that this is zero
x2.compute() - cxp.compute()

x2 += 1
# this is now all 1's -- I would have thought it would still be all 0's
x2.compute() - cxp.compute()

Do I have to somehow manually re-publish the dataset after editing it? I've tried persisting x2 but this did not seem to help. I also tried re-publishing x2 under the same name (myarr) but this raised an error that such a dataset already existed.

So is there anyway to update a published dataset in a way that all clients can access (ideally immediately) the update?

[martindurant]

I think this is expected behaviour: the in-place operation really changes what the x2 variable points to, as opposed to changing the value of the dataset which was previously persisted/published.

[funnydevnull]

@martindurant thanks for the quick response. Is there any way to implement the desired behavior? I could "unpublish" the dataset and publish a new version but I expect that other clients will not see the new version? My main goal is to have one client be able to change a shared dataset and have others see it without having to re "get_dataset". I could implement this using an external tracking mechanism (e.g. an update timestamp in a shared memory cache like redis/memcache) but this seems quite brittle (and cumbersome) so a pure dask/distributed solution would be much nicer.

[martindurant]

I don't think there is a way to change the value of some dask variable in another client automatically. Dask keys expect to refer to static values - you cannot expect to change the value of an existing future by submitting new work with the same key, for example.
Even if you had a distributed variable (generally meant for small amounts of data), you still need to get the value at some point.

[funnydevnull]

Thanks, I didn't know about variable. This might provide an alternative way of implementing this via a shared future. My problem is that I will have several quite large arrays and I want to access them in a distributed application without copying them to many workers/clients (its even possible an individual array will be larger than any clients memory). The clients are themselves distributed jobs in another task scheduler (celery).

The publish/get pattern seems to address this very nicely so long as the arrays are not updated. I suppose there are two possible solutions to updating the dataset:

• Serialize write access to the dataset using an external mechanism (a lock in a cache shared by all clients) and then if a client updates the dataset it unpublish/publishes it and sets a dirty flag in the cache prompting all other clients to get_dataset again. Any reason this won't work?

• Use a shared future for the current state of the array using a variable. Any updates to the array are written to that future directly.

Alternatively if this use case is too far out of the dask data access pattern I can just look for an alternative solution. This seems like something one should be able to do with dask but its clear I'm trying to access the data in a non-standard way.