issue_comments

Hi All,

Thanks for all of your great work, support, and discussion on these and other pages. I very much appreciate it as I am working with Xarray and Zarr quite a lot for large geospatial data storage and manipulation.

I wanted to add a note to this discussion that I have had success using Zarr's built-in ProcessSynchornizer (which relies on the fasteners package). This provides a pretty easy and clean implementation of file locks as long as you can provide a file system that is shared across any and all process that might try to access the Zarr file. For me, that means using an AWS EFS mount, which gives me the flexibility to deploy this in a serverless context or on a more standard cloud cluster.

It does seem that providing explicit chunking rules as you have mentioned above (or using the Zarr encoding argument, which I haven't tried but I think is another option) is a great way to handle this and likely outperforms the locking approach (just a guess- would love to hear from others about this). But the locks are pretty easily implemented and seem to have helped me avoid the problems related to race conditions with Zarr.

For the sake of completeness, here is a simple example of how you might do this:

synchronizer = zarr.ProcessSynchronizer(f"/mnt/efs_mnt/tmp/mur_regional_raw_sst/zarr_locks/{bounding_box['grid_loc']}_locker.sync") compressor = zarr.Blosc(cname='zstd', clevel=3) encoding = {vname: {'compressor': compressor} for vname in current_region.data_vars} current_region.to_zarr(store=store, mode='w',encoding=encoding, consolidated=True, synchronizer = synchronizer)

I would be happy to discuss further and am very much open to critique, instruction, etc.

Coming back on this issue in order not to leave it inactive and to provide some feedback to the community.

The problem with the open_mfdataset solution was that the lazy open of a single lead time dataset was still taking 150MB in memory, leading to 150*209 = 31,35GB minimum memory requirement. When I tried with a bigger (64GB memory) machine, I was then blocked with the rechunking which was exceeding the machine's resources and making the script crash. So we ended up using a dask cluster which solved the concurrency and resources limitations.

My second use-case (https://github.com/pydata/xarray/issues/3096#issuecomment-516043946) still remains though, I am wondering if it matches the intended use of zarr and if we want to do something about it, in this case I can open a separate issue documenting it.

All in all I would say my original problem is not relevant anymore, either one can do it with open_mfdataset on a single machine as proposed by @rabernat, you just need some amount of memory (and probably much more if you need to rechunk), or you do it with a dask cluster, which is the solution we chose.

Coming back on this issue (still haven't had time to try the open_mfdataset approach), I have another use case where I would like to store different variables being indexed by the same dimension, but not all available at the same moment.

For example, I would have variables V1 and V2 indexed on dimension D1. V1 would be available at time T, and I would like to store it in my S3 bucket at this moment, but V2 would only be available at time T+1. In this case, I would like to be able to save the values of V2 at time T+1, leaving the missing V2 values filled with the fill_value specified in the metadata between T and T+1.

What actually happens is that you can append such data, but then if you want to open the resulting zarr the open_zarr function needs to be given V2 as value for its drop_variables argument, otherwise you get the error shown in my original post. However, as the open_zarr function is called when appending as well (cf. original post's error trace), and in this case you can not provide this argument, you will fail the next append attempts, thus preventing you from appending the values of V2. Your dataset is now frozen.

Am I misusing the functionality, or do you know any workaround using xarray and not coding everything myself (for optimization reasons)?

Hi @VincentDehaye. Thanks for being an early adopter! We really appreciate your feedback. I'm sorry it didn't work as expected. We are in really new territory with this feature.

I'm a bit confused about why you are using the multiprocessing module here. The recommended way of parallelizing xarray operations is via the built-in dask support. There are no guarantees that multiprocessing like you're doing will work right. When we talk about parallel append, we are always talking about dask.

Your MCVE is not especially helpful for debugging because the two key functions (make_xarray_dataset and upload_to_s3) are not shown. Could you try simplifying your example a bit? I know it is hard when cloud is involved. But try to let us see more of what is happening under the hood.

If you are creating a dataset for the first time, you probably don't want append. You want to do

python ds = xr.open_mfdataset(all_the_source_files) ds.to_zarr(s3fs_target)

If you are using a dask cluster, this will automatically parallelize everything.

Hi @rabernat, thank you for your quick answer. I edited my MCVE so that you can reproduce the error(as long as you have access to a S3 bucket). I actually forgot about open_mfdataset, that's why I was doing it this way. However in the future I would still like to be able to have standalone workers, because the bandwidth quickly becomes a bottleneck for me (both on downloading the files and uploading to the cloud) so I would like to split the tasks on different machines.

With regards to open_mfdataset(), I checked the code and realized under the hood it's only calling multiple open_dataset(). I was worried it would load the values (and not only metadata) in memory, but I checked it on one file and it apparently does not. Can you confirm this ? In this case I could probably open my whole dataset at once, which would be very convenient. After reading your issue #1385, I also need to check that my case works fine with decode_cf=False. I experienced some troubles with the append on a time dimension but found a workaround, I will probably open another issue for documenting this.