issue_comments

Thanks all for the input/reviews on this PR.

(assuming tests pass)

Yes, just tried again. I'm open to ideas but would also like to move this issue along first, if possible.

Sounds good, let's go ahead and merge this!

I'm little surprised it doesn't just work with scipy and h5netcdf -- have you tried them again recently?

Yes, just tried again. I'm open to ideas but would also like to move this issue along first, if possible.

@shoyer / @mrocklin - I think this is ready for another review. Since I asked for reviews last, I have:

• reworked the task graph to write one file / close one file rather than write all / close all.
• moved the tests to the TestDask class which seemed more appropriate.

Actually, scratch that - I just found https://github.com/pydata/xarray/pull/1811/files#r183091638 which explains what is goin on. Sorry for the noise

@shoyer - I'm getting a test failure in the h5netcdf backend that seems unrelated. Do you know if something has changed in the string handling of h5netcdf recently?

The test failures in the latest build appear to be unrelated to this PR.

I see you were already using the LocalCluster (Client). Disregard my comment on switching clusters. I seem to be getting my Github issues mixed up.

It may be good to take this offline to one of the pangeo/zarr/dask/xarray issues (there are a few).

@rsignell-usgs - This is going to require some debugging on your part but here are a few suggestions:

• try smaller chunksize
• use a LocalCluster before using the KubeCluster
• turn up dask's logging level for the scheduler/workers

and after 20 seconds or so, the process dies with this error:

looking above, I don't see any error. Your goal in the next step should be to find the error. If workers are dying, they should report that in the worker logs. If your cluster is dying, that should be reported in the notebook.

@rsignell-usgs - can you repeat this example with a single threaded scheduler. It will be slow but it should work (or return a more informative error).

@shoyer - I did my best to implement the finalize store concept you've described above. This all seems to work nicely except for h5netcdf. During the store operation, h5netcdf is raising the following error:

```Python self = <h5netcdf.core._LazyObjectLookup object at 0x122d8b748>, key = 'foo'

def __getitem__(self, key):

  if self._objects[key] is not None:

E KeyError: 'foo' ```

To elaborate a little bit on my last comment (which I submitted very quickly when my bus was arriving), the way to make dependent tasks with dask.delayed is to add dummy function arguments, e.g., ```python def finalize_store(store, write): del write # unused store.sync() store.close()

write = dask.array.store(..., compute=False) write_and_close = dask.delayed(finalize_store)(store, write) write_and_close.compute() # writes and syncs ```

Potentially some of this logic could get moved into ArrayWriter.sync()

One potential issue here is the lack of clean-up (which may be unnecessary if autoclose=True). You want to construct a single dask graph with a structure like the following: - Tasks for writing all array data (i.e., from ArrayWriter). - Tasks for calling sync() and close() on each datastore object. These should depend on the appropriate writing tasks. - A single tasks that depends on writing all datastores. This is what the delayed object returned by save_mfdataset should return.