Missing zarr chunks written directly to bucket in pa... | fsspec/gcsfs

Nobody has claimed this yet.

We have recently started writing large zarr archives directly to google buckets in parallel using fsspec/gcsfs and the new regions capability in latest xarray release. We have found some intermittent issues that appear to be related to gcsfs (they are not seen when writing locally to disk).

The most serious one is that data in some chunks are not written but the process does no fail. Frequently data are missing for the last chunk along a certain dimension in the dataset. The problem is intermittent and only a small portion of the data fail to write (normally 3% or less) so it is difficult to reproduce.

I wonder if this could perhaps have to do with the rate limit of 1000 object write / sec in google buckets being exceeded or if it is related with https://github.com/intake/filesystem_spec/issues/255.

Another symptoms we have noted - sometimes we cannot read the data in the zarr archive, we get a RuntimeError: error during blosc decompression: 0 error. In other cases the process does fail with an error such as the example tracebacks below which are similar to https://github.com/pydata/xarray/issues/4704 and https://github.com/dask/gcsfs/issues/316, but it tends to succeed when we rerun it.

Example traceback 1

    self.chunk_store.setitems({k: v for k, v in zip(ckeys, cdatas)})
  File "/usr/local/lib/python3.8/dist-packages/fsspec/mapping.py", line 110, in setitems
    self.fs.pipe(values)
  File "/usr/local/lib/python3.8/dist-packages/fsspec/asyn.py", line 121, in wrapper
    return maybe_sync(func, self, *args, **kwargs)
  File "/usr/local/lib/python3.8/dist-packages/fsspec/asyn.py", line 100, in maybe_sync
    return sync(loop, func, *args, **kwargs)
  File "/usr/local/lib/python3.8/dist-packages/fsspec/asyn.py", line 71, in sync
    raise exc.with_traceback(tb)
  File "/usr/local/lib/python3.8/dist-packages/fsspec/asyn.py", line 55, in f
    result[0] = await future
  File "/usr/local/lib/python3.8/dist-packages/fsspec/asyn.py", line 222, in _pipe
    await asyncio.gather(
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 1007, in _pipe_file
    return await simple_upload(
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 1523, in simple_upload
    j = await fs._call(
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 525, in _call
    raise e
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 507, in _call
    self.validate_response(status, contents, json, path, headers)
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 1228, in validate_response
    raise HttpError(error)
gcsfs.utils.HttpError: Required

Example traceback 2

    for key in sorted(listdir(self._store, self._path)):
  File "/usr/local/lib/python3.8/dist-packages/zarr/storage.py", line 169, in listdir
    return _listdir_from_keys(store, path)
  File "/usr/local/lib/python3.8/dist-packages/zarr/storage.py", line 151, in _listdir_from_keys
    for key in list(store.keys()):
  File "/usr/lib/python3.8/_collections_abc.py", line 702, in __len__
    return len(self._mapping)
  File "/usr/local/lib/python3.8/dist-packages/fsspec/mapping.py", line 157, in __len__
    return len(self.fs.find(self.root))
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 1085, in find
    out, _ = sync(self.loop, self._do_list_objects, path, delimiter=None)
  File "/usr/local/lib/python3.8/dist-packages/fsspec/asyn.py", line 71, in sync
    raise exc.with_traceback(tb)
  File "/usr/local/lib/python3.8/dist-packages/fsspec/asyn.py", line 55, in f
    result[0] = await future
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 641, in _do_list_objects
    page = await self._call(
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 525, in _call
    raise e
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 507, in _call
    self.validate_response(status, contents, json, path, headers)
  File "/usr/local/lib/python3.8/dist-packages/gcsfs/core.py", line 1230, in validate_response
    raise HttpError({"code": status})
gcsfs.utils.HttpError

Environment:

Dask version: dask-2020.12.0
Python version: 3.8
Operating System: Ubuntu-20.04
Install method (conda, pip, source): pip

The traceback points to fsspec.mapping setitems and gcsfs/core.py in _pipe_file and _do_list_objects. Start by reproducing parallel Zarr writes to Google Cloud Storage and inspect these paths alongside the linked fsspec and gcsfs issues. Done means repeated writes complete without missing chunks, decompression errors, or intermittent listing failures.

Written by the indexing model from the issue text.

Missing zarr chunks written directly to bucket in parallel

Description

Contributor guide

First steps

Research direction

Assessment