What happened?

Hi, I have been trying to use a custom function on the rolling reduction method, the original function tries to filter the nan values (any numpy function that I have used that handles nans generates the same problem) to later apply some simple aggregate functions, but it is killing all my workers even when the data is very small (I have 7 workers and all of them have 3 Gb of RAM).

What did you expect to happen?

I would expect less use of memory taking into account the size of the rolling window, the simplicity of the function and the amount of data used on the example.

Minimal Complete Verifiable Example

```Python import numpy as np import dask.array as da import xarray as xr import dask

def f(x, axis): # If I replace np.nansum by np.sum everything works perfectly and the amount of memory used is very small return np.nansum(x, axis=axis)

arr = xr.DataArray( dask.array.zeros( shape=(300, 30000), dtype=float, chunks=(30, 6000) ), dims=["a", "b"], coords={"a": list(range(300)), "b": list(range(30000))} )

arr.rolling(a=252).reduce(f).chunk({"a": 252}).to_zarr("/data/test/test_write", mode="w") ```

MVCE confirmation

• [X] Minimal example — the example is as focused as reasonably possible to demonstrate the underlying issue in xarray.
• [X] Complete example — the example is self-contained, including all data and the text of any traceback.
• [X] Verifiable example — the example copy & pastes into an IPython prompt or Binder notebook, returning the result.
• [X] New issue — a search of GitHub Issues suggests this is not a duplicate.
• [ ] Recent environment — the issue occurs with the latest version of xarray and its dependencies.

Relevant log output

Python KilledWorker: Attempted to run task ('nansum-overlap-sum-aggregate-sum-aggregate-e732de6ad917d5f4084b05192ca671c4', 0, 0) on 4 different workers, but all those workers died while running it. The last worker that attempt to run the task was tcp://172.18.0.2:39937. Inspecting worker logs is often a good next step to diagnose what went wrong. For more information see https://distributed.dask.org/en/stable/killed.html.

Anything else we need to know?

No response

Environment

Hi I was trying to use Xarray open_zarr with the Zarr ProcessSynchronizer class and it produces a set of errors, I don't know if those errors are produced because I don't understand the logic of the ProcessSynchronizer or is a simple bug. I have a small code which reproduces the problems, basically, if I put a different path in the Zarr ProcessSynchronizer class all the error disappear but it creates a new folder.

```python import xarray import zarr import numpy as np

arr = xarray.DataArray( data=np.array([ [1, 2, 7, 4, 5], [np.nan, 3, 5, 5, 6], [3, 3, np.nan, 5, 6], [np.nan, 3, 10, 5, 6], [np.nan, 7, 8, 5, 6], ], dtype=float), dims=['index', 'columns'], coords={'index': [0, 1, 2, 3, 4], 'columns': [0, 1, 2, 3, 4]}, )

If the synchronizer is created using another path the code will work without any error but it creates a new folder,

that is the correct way to use the process synchronizer?

synchronizer = zarr.ProcessSynchronizer('dummy_array.sync')

Using the original path produce a set of weird problems

synchronizer = zarr.ProcessSynchronizer('dummy_array')

Executing the commented code I obtain: PermissionError: [WinError 5].

arr.to_dataset(name='data').to_zarr('dummy_array', mode='w', synchronizer=synchronizer, compute=True)

arr.to_dataset(name='data').to_zarr('dummy_array', mode='w', compute=True)

If this section of the code is uncommented It will throw a different error when xarray.open_zarr being executed

a = zarr.open_array('dummy_array/data', synchronizer=synchronizer, mode='r')

PermissionError: [Errno 13] Permission denied

xarray.open_zarr('dummy_array', synchronizer=synchronizer) ```

What happened: I was trying to append new data to an existing Zarr file with a time-series dataset (a financial index) and I start to notice that sometimes it produce PermissonError or randomly appear some NaN, so I check and the problem looks like is something related to multiple threads/process trying to write the same chunk (probably the lasts that has different size).

What you expected to happen: I would like to be able to store the data perfectly or it should be sufficient if it raise a NotImplemented error in case that this kind of appends is incorrect

Minimal Complete Verifiable Example: Probably you have to run many times this code to reproduce the errors, basically, you will see the PermissonError or an increment in the number of NaNs (it should has always 0) ```python import numpy as np import pandas as pd import xarray as xr

Dummy data to recreate the problem, the 308 is because my original data had this number of dates

dates = pd.bdate_range('2017-09-05', '2018-11-27')[:308] index = xr.DataArray( data=np.random.rand(len(dates)), dims=['date'], coords={'date': np.array(dates, np.datetime64)} )

Store a slice of the index in a Zarr file (new_index) using chunks with size 30

start_date = np.datetime64('2017-09-05') end_date = np.datetime64('2018-03-13') index.loc[start_date: end_date].to_dataset( name='data' ).chunk( {'date': 30} ).to_zarr( 'new_index', mode='w' )

Append the rest of the data to the new_index Zarr file

start_date = np.datetime64('2018-03-14') end_date = np.datetime64('2018-11-27')

Sometimes this section of code can produce PermissionError, probably two or more process/threads of Dask are trying

to write at the same time in the same chunks and I suppose that last chunks that end with a different size

and is necessary to 'rewrite' them are those chunks that cause the problem.

index.loc[start_date: end_date].to_dataset( name='data' ).chunk( {'date': 30} ).to_zarr( 'new_index', append_dim='date' )

The final result can contain many nan even when there is not nan in the original dataset

this behaviour is aleatory so I suppose that is related with the aforementioned error

print(xr.open_zarr('new_index')['data'].isnull().sum().compute()) print(index.isnull().sum().compute())

```

Anything else we need to know?:

Environment:

What happened?

I'm trying to use the fsspec simplecache implementation to read and write a set of Zarr files using Xarray (and some others with Dask) but during the writing process, I always get a KeyError even using the mode="w" (The complete error is on the relevant log output).

I raised a similar issue on Dask https://github.com/dask/dask/issues/9680 but when I tried the same solution of adding the "overwrite" parameter (which I think should be not necessary because Xarray already offers the mode="w" option) through the "storage_options" parameter that the "to_zarr" method offers it raises the following error: "ValueError: store must be a string to use storage_options. Got <class 'fsspec.mapping.FSMap'>", so apparently I can not apply the same solution.

What did you expect to happen?

The "to_zarr" method should be able to store the array even using the simplecache filesystem (at least Dask can).

Minimal Complete Verifiable Example

```Python mapper = fsspec.get_mapper('error_cache_write') cache_mapper = fsspec.filesystem( "simplecache", fs=mapper.fs, cache_storage='cache/files', same_names=True ).get_mapper('error_cache_write')

arr = xr.DataArray( [1, 2, 3], coords={ "test_coord": [5,6,7] } ).to_dataset(name="data")

This erases the cache and the array itself.

cache_mapper.fs.clear_cache() cache_mapper.clear()

arr.to_zarr(cache_mapper, mode="w")

Using the storage_options

arr.to_zarr(cache_mapper, mode="w", storage_options={"overwrite": True}) ```

MVCE confirmation

• [x] Minimal example — the example is as focused as reasonably possible to demonstrate the underlying issue in xarray.
• [x] Complete example — the example is self-contained, including all data and the text of any traceback.
• [x] Verifiable example — the example copy & pastes into an IPython prompt or Binder notebook, returning the result.
• [x] New issue — a search of GitHub Issues suggests this is not a duplicate.

Relevant log output

```Python KeyError Traceback (most recent call last) Input In [9], in <cell line: 16>() 8 arr = xr.DataArray( 9 [1, 2, 3], 10 coords={ 11 "test_coord": [5,6,7] 12 } 13 ).to_dataset(name="data") 14 # cache_mapper.fs.clear_cache() 15 # cache_mapper.clear() ---> 16 arr.to_zarr(cache_mapper, mode="w")

File /opt/conda/lib/python3.9/site-packages/xarray/core/dataset.py:2081, in Dataset.to_zarr(self, store, chunk_store, mode, synchronizer, group, encoding, compute, consolidated, append_dim, region, safe_chunks, storage_options) 1971 """Write dataset contents to a zarr group. 1972 1973 Zarr chunks are determined in the following way: (...) 2077 The I/O user guide, with more details and examples. 2078 """ 2079 from ..backends.api import to_zarr -> 2081 return to_zarr( # type: ignore 2082 self, 2083 store=store, 2084 chunk_store=chunk_store, 2085 storage_options=storage_options, 2086 mode=mode, 2087 synchronizer=synchronizer, 2088 group=group, 2089 encoding=encoding, 2090 compute=compute, 2091 consolidated=consolidated, 2092 append_dim=append_dim, 2093 region=region, 2094 safe_chunks=safe_chunks, 2095 )

File /opt/conda/lib/python3.9/site-packages/xarray/backends/api.py:1657, in to_zarr(dataset, store, chunk_store, mode, synchronizer, group, encoding, compute, consolidated, append_dim, region, safe_chunks, storage_options) 1655 writer = ArrayWriter() 1656 # TODO: figure out how to properly handle unlimited_dims -> 1657 dump_to_store(dataset, zstore, writer, encoding=encoding) 1658 writes = writer.sync(compute=compute) 1660 if compute:

File /opt/conda/lib/python3.9/site-packages/xarray/backends/api.py:1277, in dump_to_store(dataset, store, writer, encoder, encoding, unlimited_dims) 1274 if encoder: 1275 variables, attrs = encoder(variables, attrs) -> 1277 store.store(variables, attrs, check_encoding, writer, unlimited_dims=unlimited_dims)

File /opt/conda/lib/python3.9/site-packages/xarray/backends/zarr.py:550, in ZarrStore.store(self, variables, attributes, check_encoding_set, writer, unlimited_dims) 548 for vn in existing_variable_names: 549 vars_with_encoding[vn] = variables[vn].copy(deep=False) --> 550 vars_with_encoding[vn].encoding = existing_vars[vn].encoding 551 vars_with_encoding, _ = self.encode(vars_with_encoding, {}) 552 variables_encoded.update(vars_with_encoding)

KeyError: 'data' ```

Anything else we need to know?

No response

Environment

• [x] Closes #6112
• [x] Tests added
• [x] Passes pre-commit run --all-files
• [x] User visible changes (including notable bug fixes) are documented in whats-new.rst
• [x] New functions/methods are listed in api.rst

What happened: I'm working with a report dataset that only has data on some specific periods of time, the problem is that when I use the forward fill method it returns me many nans even on the last cells (it's a forward fill without limit).

What you expected to happen: The array should not have nans in the last cells if it has data in any other cell or there should be a warning somewhere.

Minimal Complete Verifiable Example:

```python import xarray as xr

xr.DataArray( [1, 2, np.nan, np.nan, np.nan, np.nan], dims=['a'] ).chunk( 2 ).ffill( 'a' ).compute() ``` output: array([ 1., 2., 2., 2., nan, nan])

Anything else we need to know?: I check a little bit the internal code of Xarray for forward filling when it use dask and I think that the problem is that the algorithm makes an initial forward fill on all the blocks and then it makes a map_overlap for forward filling between chunks which in case that there is an empty chunk will not work due that it is going to take the last value of the empty chunk which is nan (hope this help). Environment: