Repository: pydata/xarray

Issue:
zarr and xarray chunking compatibility and `to_zarr` performance
I have a situation where I build large zarr arrays based on chunks which correspond to how I am reading data off a filesystem, for best I/O performance. Then I set these as variables on an xarray dataset which I want to persist to zarr, but with different chunks more optimal for querying.

One problem I ran into is that manually selecting chunks of a dataset prior to `to_zarr` results in https://github.com/pydata/xarray/blob/66be9c5db7d86ea385c3a4cd4295bfce67e3f25b/xarray/backends/zarr.py#L83

It's difficult for me to understand exactly how to select chunks manually at the dataset level which would also make this zarr "final chunk" constraint happy. I would have been satisfied however with letting zarr [choose](https://github.com/zarr-developers/zarr/blob/07fa81a5afe906fcbafc5f81a851c438493ffc13/zarr/util.py#L49) chunks for me, but could not find a way to trigger this through the xarray API short of "unchunking" it first, which would lead to loading entire variables into memory. I came up with the following hack to trigger zarr's automatic chunking despite having differently defined chunks on my xarray dataset:

```python
    # monkey patch to get zarr to ignore dask chunks and use its own heuristics
    def copy_func(f):
        g = types.FunctionType(f.__code__, f.__globals__, name=f.__name__,
                               argdefs=f.__defaults__,
                               closure=f.__closure__)
        g = functools.update_wrapper(g, f)
        g.__kwdefaults__ = f.__kwdefaults__
        return g
    orig_determine_zarr_chunks = copy_func(xr.backends.zarr._determine_zarr_chunks)
    xr.backends.zarr._determine_zarr_chunks = lambda enc_chunks, var_chunks, ndim: orig_determine_zarr_chunks(enc_chunks, None, ndim)
```
The next problem to contend with is that `da.store` between zarr stores with differing chunks between source and destination is astronomically slow. The first thing to attempt would be to rechunk the dask arrays according to the destination zarr chunks, but xarray's [consistent chunks constraint](https://github.com/pydata/xarray/blob/59ad782f29a0f4766bac7802be6650be61f018b8/xarray/core/dataset.py#L1239) blocks this strategy as far as I can tell. Once again I took the dirty hack approach and inject a rechunking on a per-variable basis during the `to_zarr` operation, as follows:

```python
    # monkey patch to make dask arrays writable with different chunks than zarr dest
    # could do without this but would have to contend with 'inconsistent chunks' on dataset
    def sync_using_zarr_copy(self, compute=True):
        if self.sources:
            import dask.array as da
            rechunked_sources = [source.rechunk(target.chunks)
                for source, target in zip(self.sources, self.targets)]
            delayed_store = da.store(rechunked_sources, self.targets,
                                     lock=self.lock, compute=compute,
                                     flush=True)
            self.sources = []
            self.targets = []
            return delayed_store
    xr.backends.common.ArrayWriter.sync = sync_using_zarr_copy
```
I may have missed something in the API that would have made this easier, or another workaround which would be less hacky, but in any case I'm wondering if this scenario could be handled elegantly in xarray.

I'm not sure if there is a plan going forward to make legal xarray chunks 100% compatible with zarr; if so that would go a fair ways in alleviating the first problem. Alternatively, perhaps the xarray API could expose some ability to adjust chunks according to zarr's liking, as well as the option of defaulting entirely to zarr's heuristics for chunking.

As for the performance issue with differing chunks, I'm not sure whether my rechunking patch could be applied without causing side-effects. Or where the right place to solve this would be-- perhaps it could be more naturally addressed within `da.store`.

Discussion:
pep8speaks: Hello @rabernat! Thanks for opening this PR. We checked the lines you've touched for [PEP 8](https://www.python.org/dev/peps/pep-0008) issues, and found:



* In the file [`xarray/backends/zarr.py`](https://github.com/pydata/xarray/blob/121819c2e1186ef6aa93e9744615e81d269ee239/xarray/backends/zarr.py):

> [Line 174:40](https://github.com/pydata/xarray/blob/121819c2e1186ef6aa93e9744615e81d269ee239/xarray/backends/zarr.py#L174): [W504](https://duckduckgo.com/?q=pep8%20W504) line break after binary operator
> [Line 191:36](https://github.com/pydata/xarray/blob/121819c2e1186ef6aa93e9744615e81d269ee239/xarray/backends/zarr.py#L191): [W504](https://duckduckgo.com/?q=pep8%20W504) line break after binary operator










rabernat: Confused about the test errors. They seem unrelated. In `test_sparse.py:test_variable_method`

```
E   TypeError: no implementation found for 'numpy.allclose' on types that implement __array_function__: [<class 'numpy.ndarray'>, <class 'sparse._coo.core.COO'>]
```
andersy005: > Confused about the test error. It seems unrelated. In `test_sparse.py:test_variable_method`
> 
> ```
> E   TypeError: no implementation found for 'numpy.allclose' on types that implement __array_function__: [<class 'numpy.ndarray'>, <class 'sparse._coo.core.COO'>]
> ```


Related to #5059, and it appears that @keewis came up with a fix for it in https://github.com/pydata/xarray/pull/5059#issuecomment-803620128
rabernat: Thanks Anderson. Fixed by rebasing. Now RTD build is failing, but there is no obvious error in the logs...
shoyer: I'm a little conflicted about dealing with `encoding['chunks']` specifically in `chunk()`:

- On one hand, it feels inconsistent for this only this single method in xarray to modify part of `encoding`. Nothing else in xarray (after CF decoding) does this. Effectively `encoding['chunks']` is now becoming a part of xarray's data model.
- On the other hand, this would absolutely fix a recurrent pain-point for users, and in that sense it's worth doing.

Maybe this isn't such a big deal in this particular case, especially if we don't think we would need to add such encoding specific logic to any other methods. But are we really sure about that -- what about cases like indexing?

I guess the other alternative to make `chunk()` and various other methods that would change chunking drop `encoding` entirely. I don't know if this would really be a better comprehensive solution (I know dropping `attrs` is much hated), but at least it's an easier mental model.
rabernat: I see you point. I guess I don't fully understand where else in the code path encoding gets dropped. Consider this example

```python
import xarray as xr
ds = xr.Dataset({'foo': ('time', [1, 1], {'dtype': 'int16'})})
ds = xr.decode_cf(ds).compute()
assert "dtype" in ds.foo.encoding
assert "dtype" not in (0.5 * ds.foo).encoding
```

Xarray knows to drop the `dtype` encoding after an arithmetic operation. How does that work? To me `.chunk` feel like a similar case: an operation that invalidates any existing encoding.
shoyer: > Xarray knows to drop the `dtype` encoding after an arithmetic operation. How does that work? To me `.chunk` feel like a similar case: an operation that invalidates any existing encoding.

To be honest, the existing convention is quite adhoc, just based on what seemed most appropriate at the time.

https://github.com/pydata/xarray/issues/1614 is most comprehensive description of the current state of things.

We were considering saying that `attrs` and `encoding` should always use the same rules, but perhaps we should be more aggressive about dropping `encoding`.
dcherian: > Xarray knows to drop the dtype encoding after an arithmetic operation. How does that work? 

There's a subtle difference. It drops all of `.encoding` not `dtype` specifically.

@shoyer's point about indexing changing chunking is a good one too. Perhaps a kwarg in `to_zarr` like `ignore_encoding_chunks`?
rabernat: > Perhaps a kwarg in `to_zarr` like `ignore_encoding_chunks`?

I would argue that this is unnecessary. If you want to explicitly drop encoding, just `del da.encoding['chunks']` before writing. But most users don't figure out that they should do this, because the default behavior is counterintuitive.

The problem here is with the default behavior of propagating chunk encoding through computations when it no longer makes sense. My example with the `dtype` encoding illustrates that we already drop encoding on certain operations, so it's not unprecedented. It's more of an implementation question: where and how to do the dropping.

FWIW, I would also favor dropping `encoding['chunks']` after indexing, coarsening, interpolating, etc. Basically anything that changes the array shape or chunk structure.
shoyer: > FWIW, I would also favor dropping `encoding['chunks']` after indexing, coarsening, interpolating, etc. Basically anything that changes the array shape or chunk structure.

We already drop all of `encoding` after indexing. My guess is that we do the same for coarsening and interpolations as well (though I haven't checked).
aurghs: Perhaps we could remove also ``overwrite_encoded_chunks``, it shouldn't be any more necessary.

> The problem here is with the default behavior of propagating chunk encoding through computations when it no longer makes sense. My example with the `dtype` encoding illustrates that we already drop encoding on certain operations, so it's not unprecedented. It's more of an implementation question: where and how to do the dropping.

Maybe, in this case, the right moment to do the dropping is at reading the data. If the user doesn't re-chunk the data, the encoded chunks information is duplicated in dask-chunks.  
We could add a key in ``to_zarr``interface: ``ds.to_zarr(..., chunks=...)`` for specifying a different chunk to be used for writing the data.


rabernat: I appreciate the discussion on this PR. Does anyone have a concrete suggestion of what to do?

If we are not in agreement about the encoding stuff, perhaps I should remove that and just move forward with the `safe_chunks` part of this PR?
rabernat: In today's dev call, we proposed to handle encoding in `chunk` the same way we handle it in indexing: by deleting all encoding. 

The problem is, I can't figure out where this happens. Can someone point me to the place in the code where indexing operations delete encoding?

A related question: I discovered this encoding option `preferred_chunk`, which is treated specially:
https://github.com/pydata/xarray/blob/57a4479fcd3ebc579cf00e0d6bf85007eda44b56/xarray/core/dataset.py#L396

Should the Zarr backend be setting this?

aurghs: > Should the Zarr backend be setting this?

Yes.  We decide to separate the `chunks` and the `preferred_chunks`:
- The `preferred_chunks` is used by the backend to define the deafult chunks to be used by xarray. 
- The `chunks` are the on-disk chunks.

They are not necessarily the same. 
Maybe we can drop the `preferred_chunks` can be dropped after they are used. 


aurghs: `rechunk` is used always when you open a data with dask, even if you are using the default chunking. So in this way, you will drop the encoding always when dask is used (≈ always).
dcherian: > The problem is, I can't figure out where this happens. 

Replace `self._encoding` with `None` here? https://github.com/pydata/xarray/blob/ddc352faa6de91f266a1749773d08ae8d6f09683/xarray/core/variable.py#L1084
rabernat: > Replace `self._encoding` with `None` here?

Thanks! Yeah that's what I had in mind. But I was wondering if there was an example of doing that it else I could copy.

In any case, I'll give it a try now.
rabernat: A just pushed a new commit which deletes all encoding after chunk. But as you will see when the CI finishes, this leads to a lot of test failures. For example:

```
=============================================================================== FAILURES ================================================================================
____________________________________________________ TestNetCDF4ViaDaskData.test_roundtrip_string_encoded_characters ____________________________________________________

self = <xarray.tests.test_backends.TestNetCDF4ViaDaskData object at 0x18cba4c40>

    def test_roundtrip_string_encoded_characters(self):
        expected = Dataset({"x": ("t", ["ab", "cdef"])})
        expected["x"].encoding["dtype"] = "S1"
        with self.roundtrip(expected) as actual:
            assert_identical(expected, actual)
>           assert actual["x"].encoding["_Encoding"] == "utf-8"
E           KeyError: '_Encoding'

/Users/rpa/Code/xarray/xarray/tests/test_backends.py:485: KeyError
```

Why is `chunk` getting called here? Does it actually get called every time we load a dataset with chunks? If so, we will need a more sophisticated solution.
aurghs: >  Does it actually get called every time we load a dataset with chunks? 
Yes


rabernat: So any ideas how to proceed? 🧐
shoyer: Hmm. I would also be happy with explicitly deleting `chunks` from encoding for now. It's not adding a lot of technical debt.

In the long term, the whole handling of encoding should be revisited, e.g., see https://github.com/pydata/xarray/issues/5082
shoyer: > Why is `chunk` getting called here? Does it actually get called every time we load a dataset with chunks? If so, we will need a more sophisticated solution.

This happens specifically on this line:
https://github.com/pydata/xarray/blob/ddc352faa6de91f266a1749773d08ae8d6f09683/xarray/core/dataset.py#L438

So perhaps it would make sense to copy `encoding` specifically in this case, e.g.,
```python
        new_var = var.chunk(chunks, name=name2, lock=lock)
        new_var.encoding = var.encoding
```
aurghs: 
> ```python
>         new_var = var.chunk(chunks, name=name2, lock=lock)
>         new_var.encoding = var.encoding
> ```
Here you are modifying `_maybe_chunk`, but `_maybe_chunk` is also used in `Dataset.chunks`. 
Probably would be better to change `api.py`, here:
https://github.com/pydata/xarray/blob/ddc352faa6de91f266a1749773d08ae8d6f09683/xarray/backends/api.py#L296-L307

But maybe also in this case we want to drop `encoding["chunks"]` if they are not compatible with dask ones. We could use
https://github.com/pydata/xarray/blob/ddc352faa6de91f266a1749773d08ae8d6f09683/xarray/core/dataset.py#L363
but some small changes are needed.
aurghs: > Hmm. I would also be happy with explicitly deleting `chunks` from encoding for now. It's not adding a lot of technical debt.
 
I see two reasons for keeping it:
- We should be able to read and write the data with the same structure on file.
- We the user may be interested in this information.

But it seems to me that having two different definitions of chunks (dask one and encoded one), is not very intuitive and it's not easy to define a clear default in writing. 
rabernat: > But it seems to me that having two different definitions of chunks (dask one and encoded one), is not very intuitive and it's not easy to define a clear default in writing.

My use for `encoding.chunks` is to tell Zarr what chunks to use on disk.

rabernat: We seem to be unable to resolve the complexities around chunk encoding. I propose to remove this from the PR and reduce the scope to just the `safe_chunks` features. @aurghs should probably be the one to tackle the chunk encoding problem; unfortunately it exceeds my understanding, and I don't have time to dig deeper at the moment. In the meantime `safe_chunks` is important for pangeo-forge forward progress.

Please give a 👍  or 👎  to this idea if you have an opinion.
rabernat: I have removed the controversial `encoding['chunks']` stuff from the PR. Now it only contains the `safe_chunks` option in `to_zarr`.

If there are no further comments on this, I think this is good to go.
rabernat: Any further feedback on this now reduced-scope PR? Merging this would be helpful for moving forward Pangeo forge.
rabernat: The pre-commit workflow is raising a blackdoc error I am not seeing in my local env

```diff
diff --git a/doc/internals/duck-arrays-integration.rst b/doc/internals/duck-arrays-integration.rst
index eb5c4d8..2bc3c1f 100644
--- a/doc/internals/duck-arrays-integration.rst
+++ b/doc/internals/duck-arrays-integration.rst
@@ -25,7 +25,7 @@ argument:
         ...
 
         def _repr_inline_(self, max_width):
-            """ format to a single line with at most max_width characters """
+            """format to a single line with at most max_width characters"""
             ...
 ```
keewis: the reason is that `black` released a new version yesterday, and since we don't pin `black` for the `blackdoc` entry we get the new version. If you run `pre-commit clean` before `pre-commit run --all-files` you should see this change locally, too. To avoid situations like these we could start pinning `black` in the `blackdoc` entry (and run a script to synchronize this with the `black` entry on autoupdate).
rabernat: I think this PR has received a very thorough review. I would be pleased if someone from @pydata/xarray would merge it soon.
dcherian: Thanks @rabernat 

Patch:
diff --git a/doc/internals/duck-arrays-integration.rst b/doc/internals/duck-arrays-integration.rst
index eb5c4d85..2bc3c1f9 100644
--- a/doc/internals/duck-arrays-integration.rst
+++ b/doc/internals/duck-arrays-integration.rst
@@ -25,7 +25,7 @@ argument:
         ...
 
         def _repr_inline_(self, max_width):
-            """ format to a single line with at most max_width characters """
+            """format to a single line with at most max_width characters"""
             ...
 
         ...
diff --git a/doc/user-guide/io.rst b/doc/user-guide/io.rst
index 41cb53e4..93fc1421 100644
--- a/doc/user-guide/io.rst
+++ b/doc/user-guide/io.rst
@@ -837,11 +837,6 @@ Xarray's Zarr backend allows xarray to leverage these capabilities, including
 the ability to store and analyze datasets far too large fit onto disk
 (particularly :ref:`in combination with dask <dask>`).
 
-.. warning::
-
-    Zarr support is still an experimental feature. Please report any bugs or
-    unexepected behavior via github issues.
-
 Xarray can't open just any zarr dataset, because xarray requires special
 metadata (attributes) describing the dataset dimensions and coordinates.
 At this time, xarray can only open zarr datasets that have been written by
diff --git a/doc/whats-new.rst b/doc/whats-new.rst
index 72d67f35..ed05f92b 100644
--- a/doc/whats-new.rst
+++ b/doc/whats-new.rst
@@ -23,6 +23,9 @@ v0.17.1 (unreleased)
 New Features
 ~~~~~~~~~~~~
 
+- Add ``safe_chunks`` option to :py:meth:`Dataset.to_zarr` which allows overriding
+  checks made to ensure Dask and Zarr chunk compatibility (:issue:`5056`).
+  By `Ryan Abernathey <https://github.com/rabernat>`_
 - Add :py:meth:`Dataset.query` and :py:meth:`DataArray.query` which enable indexing
   of datasets and data arrays by evaluating query expressions against the values of the
   data variables (:pull:`4984`). By `Alistair Miles <https://github.com/alimanfoo>`_.
diff --git a/xarray/backends/api.py b/xarray/backends/api.py
index 202d7c2e..4100d177 100644
--- a/xarray/backends/api.py
+++ b/xarray/backends/api.py
@@ -1365,6 +1365,7 @@ def to_zarr(
     consolidated: bool = False,
     append_dim: Hashable = None,
     region: Mapping[str, slice] = None,
+    safe_chunks: bool = True,
 ):
     """This function creates an appropriate datastore for writing a dataset to
     a zarr ztore
@@ -1419,6 +1420,7 @@ def to_zarr(
             consolidated=consolidated,
             region=region,
             encoding=encoding,
+            # do we need to pass safe_chunks through here?
         )
 
     zstore = backends.ZarrStore.open_group(
@@ -1430,6 +1432,7 @@ def to_zarr(
         chunk_store=chunk_store,
         append_dim=append_dim,
         write_region=region,
+        safe_chunks=safe_chunks,
     )
     writer = ArrayWriter()
     # TODO: figure out how to properly handle unlimited_dims
diff --git a/xarray/backends/zarr.py b/xarray/backends/zarr.py
index 352f4971..a4086eac 100644
--- a/xarray/backends/zarr.py
+++ b/xarray/backends/zarr.py
@@ -83,7 +83,7 @@ class ZarrArrayWrapper(BackendArray):
         # could possibly have a work-around for 0d data here
 
 
-def _determine_zarr_chunks(enc_chunks, var_chunks, ndim, name):
+def _determine_zarr_chunks(enc_chunks, var_chunks, ndim, name, safe_chunks):
     """
     Given encoding chunks (possibly None) and variable chunks (possibly None)
     """
@@ -133,7 +133,7 @@ def _determine_zarr_chunks(enc_chunks, var_chunks, ndim, name):
 
     if len(enc_chunks_tuple) != ndim:
         # throw away encoding chunks, start over
-        return _determine_zarr_chunks(None, var_chunks, ndim, name)
+        return _determine_zarr_chunks(None, var_chunks, ndim, name, safe_chunks)
 
     for x in enc_chunks_tuple:
         if not isinstance(x, int):
@@ -164,24 +164,32 @@ def _determine_zarr_chunks(enc_chunks, var_chunks, ndim, name):
                 continue
             for dchunk in dchunks[:-1]:
                 if dchunk % zchunk:
-                    raise NotImplementedError(
+                    base_error = (
                         f"Specified zarr chunks encoding['chunks']={enc_chunks_tuple!r} for "
                         f"variable named {name!r} would overlap multiple dask chunks {var_chunks!r}. "
-                        "This is not implemented in xarray yet. "
-                        "Consider either rechunking using `chunk()` or instead deleting "
-                        "or modifying `encoding['chunks']`."
+                        f"Writing this array in parallel with dask could lead to corrupted data."
                     )
+                    if safe_chunks:
+                        raise NotImplementedError(
+                            base_error
+                            + " Consider either rechunking using `chunk()`, deleting "
+                            "or modifying `encoding['chunks']`, or specify `safe_chunks=False`."
+                        )
             if dchunks[-1] > zchunk:
-                raise ValueError(
+                base_error = (
                     "Final chunk of Zarr array must be the same size or "
                     "smaller than the first. "
                     f"Specified Zarr chunk encoding['chunks']={enc_chunks_tuple}, "
                     f"for variable named {name!r} "
-                    f"but {dchunks} in the variable's Dask chunks {var_chunks} is "
+                    f"but {dchunks} in the variable's Dask chunks {var_chunks} are "
                     "incompatible with this encoding. "
-                    "Consider either rechunking using `chunk()` or instead deleting "
-                    "or modifying `encoding['chunks']`."
                 )
+                if safe_chunks:
+                    raise NotImplementedError(
+                        base_error
+                        + " Consider either rechunking using `chunk()`, deleting "
+                        "or modifying `encoding['chunks']`, or specify `safe_chunks=False`."
+                    )
         return enc_chunks_tuple
 
     raise AssertionError("We should never get here. Function logic must be wrong.")
@@ -203,7 +211,9 @@ def _get_zarr_dims_and_attrs(zarr_obj, dimension_key):
     return dimensions, attributes
 
 
-def extract_zarr_variable_encoding(variable, raise_on_invalid=False, name=None):
+def extract_zarr_variable_encoding(
+    variable, raise_on_invalid=False, name=None, safe_chunks=True
+):
     """
     Extract zarr encoding dictionary from xarray Variable
 
@@ -233,7 +243,7 @@ def extract_zarr_variable_encoding(variable, raise_on_invalid=False, name=None):
                 del encoding[k]
 
     chunks = _determine_zarr_chunks(
-        encoding.get("chunks"), variable.chunks, variable.ndim, name
+        encoding.get("chunks"), variable.chunks, variable.ndim, name, safe_chunks
     )
     encoding["chunks"] = chunks
     return encoding
@@ -285,6 +295,7 @@ class ZarrStore(AbstractWritableDataStore):
         "_read_only",
         "_synchronizer",
         "_write_region",
+        "_safe_chunks",
     )
 
     @classmethod
@@ -300,6 +311,7 @@ class ZarrStore(AbstractWritableDataStore):
         storage_options=None,
         append_dim=None,
         write_region=None,
+        safe_chunks=True,
     ):
 
         # zarr doesn't support pathlib.Path objects yet. zarr-python#601
@@ -323,10 +335,17 @@ class ZarrStore(AbstractWritableDataStore):
             zarr_group = zarr.open_consolidated(store, **open_kwargs)
         else:
             zarr_group = zarr.open_group(store, **open_kwargs)
-        return cls(zarr_group, consolidate_on_close, append_dim, write_region)
+        return cls(
+            zarr_group, consolidate_on_close, append_dim, write_region, safe_chunks
+        )
 
     def __init__(
-        self, zarr_group, consolidate_on_close=False, append_dim=None, write_region=None
+        self,
+        zarr_group,
+        consolidate_on_close=False,
+        append_dim=None,
+        write_region=None,
+        safe_chunks=True,
     ):
         self.ds = zarr_group
         self._read_only = self.ds.read_only
@@ -335,6 +354,7 @@ class ZarrStore(AbstractWritableDataStore):
         self._consolidate_on_close = consolidate_on_close
         self._append_dim = append_dim
         self._write_region = write_region
+        self._safe_chunks = safe_chunks
 
     def open_store_variable(self, name, zarr_array):
         data = indexing.LazilyIndexedArray(ZarrArrayWrapper(name, self))
@@ -497,7 +517,7 @@ class ZarrStore(AbstractWritableDataStore):
             else:
                 # new variable
                 encoding = extract_zarr_variable_encoding(
-                    v, raise_on_invalid=check, name=vn
+                    v, raise_on_invalid=check, name=vn, safe_chunks=self._safe_chunks
                 )
                 encoded_attrs = {}
                 # the magic for storing the hidden dimension data
diff --git a/xarray/core/dataset.py b/xarray/core/dataset.py
index 6638834f..4c1b3b0d 100644
--- a/xarray/core/dataset.py
+++ b/xarray/core/dataset.py
@@ -1776,12 +1776,22 @@ class Dataset(DataWithCoords, DatasetArithmetic, Mapping):
         consolidated: bool = False,
         append_dim: Hashable = None,
         region: Mapping[str, slice] = None,
+        safe_chunks: bool = True,
     ) -> "ZarrStore":
         """Write dataset contents to a zarr group.
 
-        .. note:: Experimental
-                  The Zarr backend is new and experimental. Please report any
-                  unexpected behavior via github issues.
+        Zarr chunks are determined in the following way:
+
+        - From the ``chunks`` attribute in each variable's ``encoding``
+        - If the variable is a Dask array, from the dask chunks
+        - If neither Dask chunks nor encoding chunks are present, chunks will
+          be determined automatically by Zarr
+        - If both Dask chunks and encoding chunks are present, encoding chunks
+          will be used, provided that there is a many-to-one relationship between
+          encoding chunks and dask chunks (i.e. Dask chunks are bigger than and
+          evenly divide encoding chunks); otherwise raise a ``ValueError``.
+          This restriction ensures that no synchronization / locks are required
+          when writing. To disable this restriction, use ``safe_chunks=False``.
 
         Parameters
         ----------
@@ -1833,6 +1843,13 @@ class Dataset(DataWithCoords, DatasetArithmetic, Mapping):
               in with ``region``, use a separate call to ``to_zarr()`` with
               ``compute=False``. See "Appending to existing Zarr stores" in
               the reference documentation for full details.
+        safe_chunks : bool, optional
+            If True, only allow writes to when there is a many-to-one relationship
+            between Zarr chunks (specified in encoding) and Dask chunks.
+            Set False to override this restriction; however, data may become corrupted
+            if Zarr arrays are written in parallel. This option may be useful in combination
+            with ``compute=False`` to initialize a Zarr from an existing
+            Dataset with aribtrary chunk structure.
 
         References
         ----------
@@ -1869,6 +1886,7 @@ class Dataset(DataWithCoords, DatasetArithmetic, Mapping):
             consolidated=consolidated,
             append_dim=append_dim,
             region=region,
+            safe_chunks=safe_chunks,
         )
 
     def __repr__(self) -> str:


Test patch:
diff --git a/xarray/tests/test_backends.py b/xarray/tests/test_backends.py
index be5aa60b..112bcada 100644
--- a/xarray/tests/test_backends.py
+++ b/xarray/tests/test_backends.py
@@ -1871,14 +1871,21 @@ class ZarrBase(CFEncodedBase):
             with self.roundtrip(badenc) as actual:
                 pass
 
+        # unless...
+        with self.roundtrip(badenc, save_kwargs={"safe_chunks": False}) as actual:
+            # don't actually check equality because the data could be corrupted
+            pass
+
         badenc.var1.encoding["chunks"] = (2,)
-        with pytest.raises(ValueError, match=r"Specified Zarr chunk encoding"):
+        with pytest.raises(NotImplementedError, match=r"Specified Zarr chunk encoding"):
             with self.roundtrip(badenc) as actual:
                 pass
 
         badenc = badenc.chunk({"x": (3, 3, 6)})
         badenc.var1.encoding["chunks"] = (3,)
-        with pytest.raises(ValueError, match=r"incompatible with this encoding"):
+        with pytest.raises(
+            NotImplementedError, match=r"incompatible with this encoding"
+        ):
             with self.roundtrip(badenc) as actual:
                 pass
 
@@ -1901,9 +1908,13 @@ class ZarrBase(CFEncodedBase):
         # TODO: remove this failure once syncronized overlapping writes are
         # supported by xarray
         ds_chunk4["var1"].encoding.update({"chunks": 5})
-        with pytest.raises(NotImplementedError):
+        with pytest.raises(NotImplementedError, match=r"named 'var1' would overlap"):
             with self.roundtrip(ds_chunk4) as actual:
                 pass
+        # override option
+        with self.roundtrip(ds_chunk4, save_kwargs={"safe_chunks": False}) as actual:
+            # don't actually check equality because the data could be corrupted
+            pass
 
     def test_hidden_zarr_keys(self):
         expected = create_test_data()