New inline_array kwarg for open_dataset

ci/requirements/min-all-deps.yml

@@ -11,31 +11,31 @@ dependencies:
   - boto3=1.13
   - bottleneck=1.3
   # cartopy 0.18 conflicts with pynio
-  - cartopy=0.17
+  - cartopy=0.19
   - cdms2=3.1
   - cfgrib=0.9
-  - cftime=1.2
+  - cftime=1.4
   - coveralls
-  - dask-core=2.30
-  - distributed=2.30
-  - h5netcdf=0.8
+  - dask-core=2021.04
+  - distributed=2021.04
+  - h5netcdf=0.11
   - h5py=2.10
   # hdf5 1.12 conflicts with h5py=2.10
   - hdf5=1.10
   - hypothesis
   - iris=2.4
   - lxml=4.6  # Optional dep of pydap
-  - matplotlib-base=3.3
+  - matplotlib-base=3.4
   - nc-time-axis=1.2
   # netcdf follows a 1.major.minor[.patch] convention
   # (see https://github.com/Unidata/netcdf4-python/issues/1090)
   # bumping the netCDF4 version is currently blocked by #4491
   - netcdf4=1.5.3
-  - numba=0.51
-  - numpy=1.18
+  - numba=0.53
+  - numpy=1.19
   - packaging=20.0
-  - pandas=1.1
-  - pint=0.16
+  - pandas=1.2
+  - pint=0.17
   - pip
   - pseudonetcdf=3.1
   - pydap=3.2
@@ -44,12 +44,12 @@ dependencies:
   - pytest-cov
   - pytest-env
   - pytest-xdist
-  - rasterio=1.1
-  - scipy=1.5
+  - rasterio=1.2
+  - scipy=1.6
   - seaborn=0.11
-  - sparse=0.11
+  - sparse=0.12
   - toolz=0.11
   - typing_extensions=3.7
-  - zarr=2.5
+  - zarr=2.8
   - pip:
     - numbagg==0.1

doc/getting-started-guide/installing.rst

@@ -9,7 +9,7 @@ Required dependencies
 - Python (3.8 or later)
 - `numpy <https://www.numpy.org/>`__ (1.18 or later)
 - `packaging <https://packaging.pypa.io/en/latest/#>`__ (20.0 or later)
-- `pandas <https://pandas.pydata.org/>`__ (1.1 or later)
+- `pandas <https://pandas.pydata.org/>`__ (1.2 or later)
 
 .. _optional-dependencies:
 

setup.cfg

@@ -76,7 +76,7 @@ include_package_data = True
 python_requires = >=3.8
 install_requires =
     numpy >= 1.18
-    pandas >= 1.1
+    pandas >= 1.2
     packaging >= 20.0
 
 [options.extras_require]

xarray/backends/api.py

@@ -274,6 +274,7 @@ def _chunk_ds(
     engine,
     chunks,
     overwrite_encoded_chunks,
+    inline_array,
     **extra_tokens,
 ):
     from dask.base import tokenize
@@ -292,6 +293,7 @@ def _chunk_ds(
             overwrite_encoded_chunks=overwrite_encoded_chunks,
             name_prefix=name_prefix,
             token=token,
+            inline_array=inline_array,
         )
     return backend_ds._replace(variables)
 
@@ -303,6 +305,7 @@ def _dataset_from_backend_dataset(
     chunks,
     cache,
     overwrite_encoded_chunks,
+    inline_array,
     **extra_tokens,
 ):
     if not isinstance(chunks, (int, dict)) and chunks not in {None, "auto"}:
@@ -320,6 +323,7 @@ def _dataset_from_backend_dataset(
             engine,
             chunks,
             overwrite_encoded_chunks,
+            inline_array,
             **extra_tokens,
         )
 
@@ -444,6 +448,11 @@ def open_dataset(
           appropriate locks are chosen to safely read and write files with the
           currently active dask scheduler. Supported by "netcdf4", "h5netcdf",
           "scipy", "pynio", "pseudonetcdf", "cfgrib".
+        - 'inline_array': How to include the array in the dask task graph. By
+          default(``inline_array=False``) the array is included in a task by
+          itself, and each chunk refers to that task by its key. With
+          ``inline_array=True``, Dask will instead inline the array directly
+          in the values of the task graph. See `dask.array.from_array()`.
 
         See engine open function for kwargs accepted by each specific engine.
 
@@ -492,6 +501,7 @@ def open_dataset(
     )
 
     overwrite_encoded_chunks = kwargs.pop("overwrite_encoded_chunks", None)
+    inline_array = kwargs.pop("inline_array", False)
     backend_ds = backend.open_dataset(
         filename_or_obj,
         drop_variables=drop_variables,
@@ -505,6 +515,7 @@ def open_dataset(
         chunks,
         cache,
         overwrite_encoded_chunks,
+        inline_array,
         drop_variables=drop_variables,
         **decoders,
         **kwargs,
@@ -627,6 +638,11 @@ def open_dataarray(
           appropriate locks are chosen to safely read and write files with the
           currently active dask scheduler. Supported by "netcdf4", "h5netcdf",
           "scipy", "pynio", "pseudonetcdf", "cfgrib".
+        - 'inline_array': How to include the array in the dask task graph. By
+          default(``inline_array=False``) the array is included in a task by
+          itself, and each chunk refers to that task by its key. With
+          ``inline_array=True``, Dask will instead inline the array directly
+          in the values of the task graph. See `dask.array.from_array()`.
 
         See engine open function for kwargs accepted by each specific engine.
 

xarray/core/dataarray.py

@@ -1113,6 +1113,7 @@ def chunk(
         name_prefix: str = "xarray-",
         token: str = None,
         lock: bool = False,
+        inline_array: bool = False,
         **chunks_kwargs: Any,
     ) -> DataArray:
         """Coerce this array's data into a dask arrays with the given chunks.
@@ -1137,13 +1138,23 @@ def chunk(
         lock : optional
             Passed on to :py:func:`dask.array.from_array`, if the array is not
             already as dask array.
+        inline_array: optional
+            Passed on to :py:func:`dask.array.from_array`, if the array is not
+            already as dask array.
         **chunks_kwargs : {dim: chunks, ...}, optional
             The keyword arguments form of ``chunks``.
             One of chunks or chunks_kwargs must be provided.
 
         Returns
         -------
         chunked : xarray.DataArray
+
+        See Also
+        --------
+        DataArray.chunks
+        DataArray.chunksizes
+        xarray.unify_chunks
+        dask.array.from_array
         """
         if chunks is None:
             warnings.warn(
@@ -1162,7 +1173,11 @@ def chunk(
             chunks = either_dict_or_kwargs(chunks, chunks_kwargs, "chunk")
 
         ds = self._to_temp_dataset().chunk(
-            chunks, name_prefix=name_prefix, token=token, lock=lock
+            chunks,
+            name_prefix=name_prefix,
+            token=token,
+            lock=lock,
+            inline_array=inline_array,
         )
         return self._from_temp_dataset(ds)
 

xarray/core/dataset.py

@@ -239,6 +239,7 @@ def _maybe_chunk(
     lock=None,
     name_prefix="xarray-",
     overwrite_encoded_chunks=False,
+    inline_array=False,
 ):
     from dask.base import tokenize
 
@@ -250,7 +251,7 @@ def _maybe_chunk(
         # subtle bugs result otherwise. see GH3350
         token2 = tokenize(name, token if token else var._data, chunks)
         name2 = f"{name_prefix}{name}-{token2}"
-        var = var.chunk(chunks, name=name2, lock=lock)
+        var = var.chunk(chunks, name=name2, lock=lock, inline_array=inline_array)
 
         if overwrite_encoded_chunks and var.chunks is not None:
             var.encoding["chunks"] = tuple(x[0] for x in var.chunks)
@@ -1994,6 +1995,7 @@ def chunk(
         name_prefix: str = "xarray-",
         token: str = None,
         lock: bool = False,
+        inline_array: bool = False,
         **chunks_kwargs: Any,
     ) -> Dataset:
         """Coerce all arrays in this dataset into dask arrays with the given
@@ -2018,6 +2020,9 @@ def chunk(
         lock : optional
             Passed on to :py:func:`dask.array.from_array`, if the array is not
             already as dask array.
+        inline_array: optional
+            Passed on to :py:func:`dask.array.from_array`, if the array is not
+            already as dask array.
         **chunks_kwargs : {dim: chunks, ...}, optional
             The keyword arguments form of ``chunks``.
             One of chunks or chunks_kwargs must be provided
@@ -2031,6 +2036,7 @@ def chunk(
         Dataset.chunks
         Dataset.chunksizes
         xarray.unify_chunks
+        dask.array.from_array
         """
         if chunks is None and chunks_kwargs is None:
             warnings.warn(

xarray/core/variable.py

@@ -1023,6 +1023,7 @@ def chunk(
         ) = {},
         name: str = None,
         lock: bool = False,
+        inline_array: bool = False,
         **chunks_kwargs: Any,
     ) -> Variable:
         """Coerce this array's data into a dask array with the given chunks.
@@ -1046,13 +1047,23 @@ def chunk(
         lock : optional
             Passed on to :py:func:`dask.array.from_array`, if the array is not
             already as dask array.
+        inline_array: optional
+            Passed on to :py:func:`dask.array.from_array`, if the array is not
+            already as dask array.
         **chunks_kwargs : {dim: chunks, ...}, optional
             The keyword arguments form of ``chunks``.
             One of chunks or chunks_kwargs must be provided.
 
         Returns
         -------
         chunked : xarray.Variable
+
+        See Also
+        --------
+        Variable.chunks
+        Variable.chunksizes
+        xarray.unify_chunks
+        dask.array.from_array
         """
         import dask.array as da
 
@@ -1098,7 +1109,9 @@ def chunk(
             if utils.is_dict_like(chunks):
                 chunks = tuple(chunks.get(n, s) for n, s in enumerate(self.shape))
 
-            data = da.from_array(data, chunks, name=name, lock=lock, **kwargs)
+            data = da.from_array(
+                data, chunks, name=name, lock=lock, inline_array=inline_array, **kwargs
+            )
 
         return self._replace(data=data)
 
@@ -2710,7 +2723,7 @@ def values(self, values):
             f"Please use DataArray.assign_coords, Dataset.assign_coords or Dataset.assign as appropriate."
         )
 
-    def chunk(self, chunks={}, name=None, lock=False):
+    def chunk(self, chunks={}, name=None, lock=False, inline_array=False):
         # Dummy - do not chunk. This method is invoked e.g. by Dataset.chunk()
         return self.copy(deep=False)