issue_comments

cc @ivirshup @joshmoore (who may be interested in this as well)

FWIW this sounds similar to what h5pickle does. Maybe it is worth improving that package with whatever logic Xarray has?

Wanted to note issue ( https://github.com/carbonplan/ndpyramid/issues/10 ) here, which may be of interest to people here.

Also we are thinking about a Dask blogpost in this space if people have thoughts on what we should include and/or are interested in being involved. Details in issue ( https://github.com/dask/dask-blog/issues/141 ).

Probably out of my depth here (so please forgive me), but one thing that might be worth looking at is Array API support, which CuPy 10+ supports and Dask is working on support for ( https://github.com/dask/dask/pull/8750 ). Believe XArray is taking some initial steps in this direction recently ( https://github.com/pydata/xarray/pull/6804 ), but could easily be misunderstanding the scope/intended usage of the changes there.

cc @rgommers (for awareness)

While it is true to use PyTorch Tensors directly, one would need the Array API implemented in PyTorch. One could use them indirectly by converting them zero-copy to CuPy arrays, which do have Array API support

@jakirkham were you thinking a reference to the dask docs for more info on optimal chunk sizing and aligning with storage?

It could make sense to refer to or if similar ideas come up here it may be worth mentioning in this change

or are you suggesting the proposed docs change is too complex?

Not at all.

I was trying to address the lack of documentation on specifying chunks within a zarr array for non-dask arrays/coordinates, but also covering the weedsy (but common) case of datasets with a mix of dask & in-memory arrays/coords like in my example. I have been frustrated by zarr stores I've written with a couple dozen array chunks and thousands of coordinate chunks for this reason, but it's definitely a gnarly topic to cover concisely :P

If there's anything you need help with or would like to discuss, please don't hesitate to raise a Zarr issue. We also enabled GH discussions over there so if that fits better feel free to use that 🙂

FWIW there's a similar doc page about chunk size in Dask that may be worth borrowing from

Would be good to keep this open

If you haven't already, would be good if those running into issues here could look over the Array API. This is still something that is being worked on, but the goal is to standardize Array APIs. If there are things missing from that, it would be good to hear about them in a new issue.

Maybe too soon to ask, but do we have a link to the video call?

@Illviljan please let us know if there's anything else needed here 🙂

Thanks Hameer! 😄

cc @pentschev (just so you are aware)

Yeah was just mentioning that since we had older version of sparse pulled while developing that PR at one point and it caused issues. Sounds like that is not the case here

Would double check that CI is pulling the latest sparse

xref: https://github.com/dask/dask/pull/7939#issuecomment-887122942

Yeah +1 for using pad instead. Had tried to get rid of map_overlap's padding and use da.pad in Dask as well ( https://github.com/dask/dask/pull/5052 ), but haven't had time to get back to that.

Well here's a blogpost on using Dask + CuPy. Maybe start there and build up to using Xarray.

Yeah Jacob and I played with this a few months back. There were some issues, but my recollection is pretty hazy. If someone gives this another try, it would be interesting to hear how things go.

Sure an upstream issue would be welcome. Thanks for unpacking that further Mark 😀

Sorry I don't know. Maybe @rabernat can advise? 🙂

Yeah this probably works as these are just JSON files. That said, IDK that we are making any attempt to ensure this works. IOW I don't think this is tested or in the spec.

Additionally IDK that we do the same decoding on nested dictionaries as would be done on a flat dictionary. Meaning non-JSON values like datetime64/timedelta64 might not be handled correctly in this case.

Could be wrong about these things. Those are just my immediate thoughts.

Congratulations! 🎉

Another approach for the split_by_chunks implementation would be...

python def split_by_chunks(a): for sl in da.core.slices_from_chunks(a.chunks): yield (sl, a[sl])

While a little bit more cumbersome to write, this could be implemented with .blocks and may be a bit more performant.

python def split_by_chunks(a): for i, sl in zip(np.ndindex(a.numblocks), da.core.slices_from_chunks(a.chunks)): yield (sl, a.blocks[i])

If the slices are not strictly needed, this could be simplified a bit more.

python def split_by_chunks(a): for i in np.ndindex(a.numblocks): yield a.blocks[i]

Admittedly slices_from_chunks is an internal utility function. Though it is unlikely to change. We could consider exposing it as part of the API if that is useful.

We could consider other things like making .blocks iterable, which could make this more friendly as well. Raised issue ( https://github.com/dask/dask/issues/5117 ) on this point.

That sounds somewhat similar to .blocks accessor in Dask Array. ( https://github.com/dask/dask/pull/3689 ) Maybe we should align on that as well?

I'm not really familiar with XArray's internals, but issue ( https://github.com/pydata/xarray/issues/2660 ) looks relevant.

What happens if you do?

python ds.to_zarr(zarr.group(zarr.ZipStore("test.zarr"))) print(xr.open_zarr(zarr.group(zarr.ZipStore("test.zarr"))))

@shoyer, have you seen da.pad?

Is this still an issue?

from what I understand da.dot implements... a limited special case of da.einsum?

Basically dot is an inner product. Certainly inner products can be formulated using Einstein notation (i.e. calling with einsum).

The question is whether the performance keeps up with that formulation. Currently it sounds like chunking causes some problems right now IIUC. However things like dot and tensordot dispatch through optimized BLAS routines. In theory einsum should do the same ( https://github.com/numpy/numpy/pull/9425 ), but the experimental data still shows a few warts. For example, matmul is implemented with einsum, but is slower than dot. ( https://github.com/numpy/numpy/issues/7569 ) ( https://github.com/numpy/numpy/issues/8957 ) Pure einsum implementations seem to perform similarly.

I ran a few more benchmarks...

What are the arrays used as input for this case?

...apparently xarray.dot on a dask backend is situationally faster than all other implementations when you are not reducing on any dimensions...

Having a little trouble following this. dot reduces one dimension from each input. Excepting if one of the inputs is 0-D (i.e. a scalar), then it is just multiplying a single scalar through an array. Is that what you are referring?

Might be worth revisiting how da.dot is implemented as well. That would be the least amount of rewriting for you and would generally be nice for Dask users. If you have not already, @crusaderky, it would be nice to raise an issue over at Dask with a straight Dask benchmark comparing Dask Array's dot and einsum.

cc @mrocklin

What is store in this case? Sorry not very familiar with how xarray does things.

So Zarr supports storing structured arrays. Maybe that’s what you are looking for, @martindurant? Would suggest using the latest 2.2.0 RC though as it fixed a few issues in this regard (particularly with NumPy 1.14).

Yep, using dask.array.store regularly with the distributed scheduler both on our cluster and in a local Docker image for testing. Am using Zarr Arrays as the targets for store to write to. Basically rechunk the data to match the chunking selected for the Zarr Array and then write out in parallel lock-free.

Our cluster uses NFS for things like one's home directory. So these are accessible across nodes. Also there are other types of storage available that are a bit faster and still remain accessible across nodes. So these work pretty well.

In case anyone is curious, PR ( https://github.com/dask/dask/pull/2980 ) contains this work. Feedback welcome.

Just to give a brief synopsis of what we are working in Dask in case it is valuable for this or other contexts, have given an overview of the relevant work below.

With Matthew's help am trying to add a keep argument to da.store. By default keep=False, which is the current behavior of da.store. If keep=True however, it returns Dask Arrays that can lazily load data written by da.store. Thus allowing the stored result to be linked to later computations before it is fully written. The compute argument of da.store affects whether to submit the storage tasks immediately (adding Futures into the resultant Dask Array) or whether to hold off until a later computation step triggers it.

This sort of functionality could be useful for a variety of situations including the one Matthew has described above. Also this could be useful for viewing partially computed results before they are totally done. Another use case could be more rapid batching of computations with many intermediate values. There is also an opportunity to re-explore caching in this context; thus, revisiting an area that many people have previously shown interest in.

Just to confirm, if writes are aligned with chunk boundaries in the destination array then no locking is required.

As a minor point to complement what Matthew and Alistair have already said, one can pretty easily rechunk beforehand so that the chunks will have a nice 1-to-1 non-overlapping mapping on disk. Not sure whether this strategy is good enough to make default. However have had no issues doing this myself. Also would expect it is better than holding one lock over the whole Zarr Array. Though there may be some strange edge cases that I have not encountered.

Given the recent turn in discussion here, might be worthwhile to share some thoughts on issue ( https://github.com/dask/dask/issues/2694 ).

Also, FYI we have python-coveralls currently. Though we don't have coveralls yet.

Please let me know if there is something actionable for me here. Looks like that is not the case. If that changes, please let me know.