def open_zarr(store, group=None, synchronizer=None, chunks='auto',
              decode_cf=True, mask_and_scale=True, decode_times=True,
              concat_characters=True, decode_coords=True,
              drop_variables=None, consolidated=False,
              overwrite_encoded_chunks=False, **kwargs):
    """Load and decode a dataset from a Zarr store.

    .. note:: Experimental
              The Zarr backend is new and experimental. Please report any
              unexpected behavior via github issues.

    The `store` object should be a valid store for a Zarr group. `store`
    variables must contain dimension metadata encoded in the
    `_ARRAY_DIMENSIONS` attribute.

    Parameters
    ----------
    store : MutableMapping or str
        A MutableMapping where a Zarr Group has been stored or a path to a
        directory in file system where a Zarr DirectoryStore has been stored.
    synchronizer : object, optional
        Array synchronizer provided to zarr
    group : str, obtional
        Group path. (a.k.a. `path` in zarr terminology.)
    chunks : int or dict or tuple or {None, 'auto'}, optional
        Chunk sizes along each dimension, e.g., ``5`` or
        ``{'x': 5, 'y': 5}``. If `chunks='auto'`, dask chunks are created
        based on the variable's zarr chunks. If `chunks=None`, zarr array
        data will lazily convert to numpy arrays upon access. This accepts
        all the chunk specifications as Dask does.
    overwrite_encoded_chunks: bool, optional
        Whether to drop the zarr chunks encoded for each variable when a
        dataset is loaded with specified chunk sizes (default: False)
    decode_cf : bool, optional
        Whether to decode these variables, assuming they were saved according
        to CF conventions.
    mask_and_scale : bool, optional
        If True, replace array values equal to `_FillValue` with NA and scale
        values according to the formula `original_values * scale_factor +
        add_offset`, where `_FillValue`, `scale_factor` and `add_offset` are
        taken from variable attributes (if they exist).  If the `_FillValue` or
        `missing_value` attribute contains multiple values a warning will be
        issued and all array values matching one of the multiple values will
        be replaced by NA.
    decode_times : bool, optional
        If True, decode times encoded in the standard NetCDF datetime format
        into datetime objects. Otherwise, leave them encoded as numbers.
    concat_characters : bool, optional
        If True, concatenate along the last dimension of character arrays to
        form string arrays. Dimensions will only be concatenated over (and
        removed) if they have no corresponding variable and if they are only
        used as the last dimension of character arrays.
    decode_coords : bool, optional
        If True, decode the 'coordinates' attribute to identify coordinates in
        the resulting dataset.
    drop_variables : string or iterable, optional
        A variable or list of variables to exclude from being parsed from the
        dataset. This may be useful to drop variables with problems or
        inconsistent values.
    consolidated : bool, optional
        Whether to open the store using zarr's consolidated metadata
        capability. Only works for stores that have already been consolidated.

    Returns
    -------
    dataset : Dataset
        The newly created dataset.

    See Also
    --------
    open_dataset

    References
    ----------
    http://zarr.readthedocs.io/
    """
    if 'auto_chunk' in kwargs:
        auto_chunk = kwargs.pop('auto_chunk')
        if auto_chunk:
            chunks = 'auto'  # maintain backwards compatibility
        else:
            chunks = None

        warnings.warn("auto_chunk is deprecated. Use chunks='auto' instead.",
                      FutureWarning, stacklevel=2)

    if kwargs:
        raise TypeError("open_zarr() got unexpected keyword arguments " +
                        ",".join(kwargs.keys()))

    if not isinstance(chunks, (int, dict)):
        if chunks != 'auto' and chunks is not None:
            raise ValueError("chunks must be an int, dict, 'auto', or None. "
                             "Instead found %s. " % chunks)

    if not decode_cf:
        mask_and_scale = False
        decode_times = False
        concat_characters = False
        decode_coords = False

    def maybe_decode_store(store, lock=False):
        ds = conventions.decode_cf(
            store, mask_and_scale=mask_and_scale, decode_times=decode_times,
            concat_characters=concat_characters, decode_coords=decode_coords,
            drop_variables=drop_variables)

        # TODO: this is where we would apply caching

        return ds

    # Zarr supports a wide range of access modes, but for now xarray either
    # reads or writes from a store, never both. For open_zarr, we only read
    mode = 'r'
    zarr_store = ZarrStore.open_group(store, mode=mode,
                                      synchronizer=synchronizer,
                                      group=group, consolidated=consolidated)
    ds = maybe_decode_store(zarr_store)

    # auto chunking needs to be here and not in ZarrStore because variable
    # chunks do not survive decode_cf
    # return trivial case
    if not chunks:
        return ds

    # adapted from Dataset.Chunk()
    if isinstance(chunks, int):
        chunks = dict.fromkeys(ds.dims, chunks)

    if isinstance(chunks, tuple) and len(chunks) == len(ds.dims):
        chunks = dict(zip(ds.dims, chunks))

    def get_chunk(name, var, chunks):
        chunk_spec = dict(zip(var.dims, var.encoding.get('chunks')))

        # Coordinate labels aren't chunked
        if var.ndim == 1 and var.dims[0] == name:
            return chunk_spec

        if chunks == 'auto':
            return chunk_spec

        for dim in var.dims:
            if dim in chunks:
                spec = chunks[dim]
                if isinstance(spec, int):
                    spec = (spec,)
                if isinstance(spec, (tuple, list)) and chunk_spec[dim]:
                    if any(s % chunk_spec[dim] for s in spec):
                        warnings.warn("Specified Dask chunks %r would "
                                      "separate Zarr chunk shape %r for "
                                      "dimension %r. This significantly "
                                      "degrades performance. Consider "
                                      "rechunking after loading instead."
                                      % (chunks[dim], chunk_spec[dim], dim),
                                      stacklevel=2)
                chunk_spec[dim] = chunks[dim]
        return chunk_spec

    def maybe_chunk(name, var, chunks):
        from dask.base import tokenize

        chunk_spec = get_chunk(name, var, chunks)

        if (var.ndim > 0) and (chunk_spec is not None):
            # does this cause any data to be read?
            token2 = tokenize(name, var._data)
            name2 = 'zarr-%s' % token2
            var = var.chunk(chunk_spec, name=name2, lock=None)
            if overwrite_encoded_chunks and var.chunks is not None:
                var.encoding['chunks'] = tuple(x[0] for x in var.chunks)
            return var
        else:
            return var

    variables = OrderedDict([(k, maybe_chunk(k, v, chunks))
                            for k, v in ds.variables.items()])
    return ds._replace_vars_and_dims(variables)