Opening zarr dataset with poor connection leads to NaN chunks

[renaudjester]

Problem

I am using xarray to open zarr datasets located in an s3 bucket.
However, it can happen that the results doesn't retrieve all the chunks and we have NaNs instead.
It is usually linked with low bandwith internet connection and asking for a lot of chunks.

More details

In our case (see code below), we started tracking the http calls to understand a bit better the problem (with http tracking software). 3 cases are possible as for the response when getting a chunk:

• 200: we get the chunk with the data
• 403: missing data, this is normal as I am dealing with ocean data so the chunks associated with the continent don't exists
• no response: there isn't even a response so the get request "fails" and we don't have the data.
  

The latter is a big problem as we have randomly empty chunks! as a user it is also very annoying to detect.

We also noticed that when using xarray.open_dataset the calls seems to be done all at the same time! Which increases the probability of NaN chunks. That's why we tried using xarray.open_mfdataset since each worker calls the get request ie the chunks, one by one.

Questions

• Why the xarray.open_dataset sends all the requests concurrently? Is it possible to control the number of requests and do some kind of rolling batch gather?
• Is there a way to raise an exception when there are no response from the server? So that at least, as users, we don't have to manually check the data.
• Any idea to solve this problem?
• Maybe this is linked to zarr library?

To reproduce

This bug is difficult to reproduce. The only way I managed to reproduce it is with a computer connected to a phone that is connected to the 3G. With this setup it happens all the time though. With a good connection and on my computer it never happens.
We have had several reports of this problem otherwise.

See the two scripts: one with open_dataset

import xarray as xr
import matplotlib.pyplot as plt
import time
import sys

import logging
logging.basicConfig(
    stream=sys.stdout,
    format="%(asctime)s | %(name)14s | %(levelname)7s | %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
    encoding="utf-8",
    level=logging.ERROR,
)
logging.getLogger("timeloop").setLevel(logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)
logging.getLogger("botocore").setLevel(logging.DEBUG)
logging.getLogger("s3fs").setLevel(logging.DEBUG)
logging.getLogger("fsspec").setLevel(logging.DEBUG)
logging.getLogger("asyncio").setLevel(logging.DEBUG)
# logging.getLogger("numba").setLevel(logging.ERROR)
logging.getLogger("s3transfer").setLevel(logging.DEBUG)

start_time = time.time()

print("Starting...")
data = xr.open_dataset("https://s3.waw3-1.cloudferro.com/mdl-arco-geo-012/arco/GLOBAL_ANALYSISFORECAST_PHY_001_024/cmems_mod_glo_phy-thetao_anfc_0.083deg_P1D-m_202211/geoChunked.zarr", 
engine = "zarr")

print("Dataset opened...")
bla = data.thetao.sel(longitude = slice(-170.037309004901026,-70.037309004901026),
               latitude=slice(-80.27257431850789,-40.27257431850789),
                time=slice("2023-03-20T00:00:00","2023-03-20T00:00:00")).sel(elevation =0,
method="nearest")


print("Plotting... ")
map = bla.isel(time=0).plot()

#map = data.isel(time=0).plot()

print("Saving image...")
plt.savefig("./bla_fast.png")

print("Total processing time:", (time.time() - start_time))

and the other one with open_mfdataset:

import xarray as xr
import matplotlib.pyplot as plt
import time
import sys
import dask

import logging
logging.basicConfig(
    stream=sys.stdout,
    format="%(asctime)s | %(name)14s | %(levelname)7s | %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
    encoding="utf-8",
    level=logging.ERROR,
)
logging.getLogger("timeloop").setLevel(logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)
logging.getLogger("botocore").setLevel(logging.DEBUG)
logging.getLogger("s3fs").setLevel(logging.DEBUG)
logging.getLogger("fsspec").setLevel(logging.DEBUG)
logging.getLogger("asyncio").setLevel(logging.DEBUG)
# logging.getLogger("numba").setLevel(logging.ERROR)
logging.getLogger("s3transfer").setLevel(logging.DEBUG)

start_time = time.time()

with dask.config.set(num_workers=2):

	print("Starting...")
	data = xr.open_mfdataset(["https://s3.waw3-1.cloudferro.com/mdl-arco-geo-012/arco/GLOBAL_ANALYSISFORECAST_PHY_001_024/cmems_mod_glo_phy-thetao_anfc_0.083deg_P1D-m_202211/geoChunked.zarr"], 
	engine = "zarr")#.thetao.sel(longitude = slice(-170.037309004901026,-70.037309004901026),
                #latitude=slice(-80.27257431850789,-40.27257431850789),
                #time=slice("2023-03-20T00:00:00","2023-03-20T00:00:00")).sel(elevation =0, 	#method="nearest")

	print("Dataset opened...")
	bla = data.thetao.sel(longitude = slice(-170.037309004901026,-70.037309004901026),
                latitude=slice(-80.27257431850789,-40.27257431850789),
                time=slice("2023-03-20T00:00:00","2023-03-20T00:00:00")).sel(elevation =0,
	method="nearest")


	print("Plotting... ")
	map = bla.isel(time=0).plot()

	#map = data.isel(time=0).plot()

	print("Saving image...")
	plt.savefig("./bla_long.png")

print("Total processing time:", (time.time() - start_time))

Expected result

or failed run

Obtained result

[welcome]

Thanks for opening your first issue here at xarray! Be sure to follow the issue template!
If you have an idea for a solution, we would really welcome a Pull Request with proposed changes.
See the Contributing Guide for more.
It may take us a while to respond here, but we really value your contribution. Contributors like you help make xarray better.
Thank you!

[dcherian]

I suspect fsspec should raise an error here, but isn't. cc @martindurant

[renaudjester]

Indeed, with the same set ups and only doing the fsspec calls consurrently leads to some unexpected error: empty response, Server disconnected and [Errno 54] Connection reset by peer (I can provide some example snippet if needed)

And @dcherian they DO raise when I leave the default fsspec default behavior (see https://filesystem-spec.readthedocs.io/en/latest/api.html?highlight=.cat#fsspec.spec.AbstractFileSystem.cat) so I actually think that it might not be an upstream issue.

So I am wondering: do you know where in the process there is an "omit" and we bypass all errors?
Is it in xarray? In zarr? In which place?

Thanks 😄

[martindurant]

So I am wondering: do you know where in the process there is an "omit" and we bypass all errors?

It is not an omit, but errors can be converted into FileNotFoundError either in fsspec's FSMap, or in zarr's FSStore (these two should do the same thing and are based on each other). zarr assumes NotFound means a file is really missing, and that the data there is all fill_value - this is the design.

In some situations it can be hard to tell the difference between something that isn't there and something that is, but won't load. This is one argument for having an extenral manifest of which files are expected - but then you need to keep this manifest in sync with reality.

[renaudjester]

I agree that it can be hard to tell the difference between something that isn't there and something that is. Though I am surprised that some errors are not propagated at all.

We made some specific zarr store to be able to retry those errors (and raise them if needed). Here is a little part of the store (I can share the rest with you).

def __contains__(self, key):
        full_key = f"{self._root_path}/{key}"
        try:
            self.client.head_object(Bucket=self._bucket, Key=full_key)
            return True
        except botocore.exceptions.ClientError as e:
            if "404" in str(e) or "403" in str(e):
                return False
            raise

You can see that we are only catching the 404 and 403 errors which correspond missing chunk. I was expecting xarray to work in the same fashion.

[dcherian]

but errors can be converted into FileNotFoundError either in fsspec's FSMap, or in zarr's FSStore

We should fix this. Connection errors should be distinguishable from a true FileNotFound error though, especially since missing chunk files are valid for Zarr.

cc @jhamman

[martindurant]

It is configurable: https://github.com/zarr-developers/zarr-python/blob/main/zarr/storage.py#L1354 (I don't know why this is broader than https://github.com/fsspec/filesystem_spec/blob/master/fsspec/mapping.py#L46 ).

[renaudjester]

Can it be configured through xarray (through open_dataset for example) directly then?

I guess it would be interesting to have the a retry policy also

[martindurant]

Yes, you should be able to put this into the storage_options within the backend_kwargs

[renaudjester]

I doesn't have any effect unfortunately, as you see

[saschahofmann]

@martindurant as renaud mentions the storage_option.exceptions doesnt seem to be passed to the ZarrStore. doing something like backen_kwargs={"exceptions": KeyError} throws an ArgumentError.

I am looking at https://github.com/pydata/xarray/blob/main/xarray/backends/zarr.py#L995 and can't really see how the storage_options or backend_kwargs would be passed to the Zarrstore?