diff --git a/doc/dask.rst b/doc/dask.rst index 672450065cb..975111cba33 100644 --- a/doc/dask.rst +++ b/doc/dask.rst @@ -1,49 +1,49 @@ .. _dask: -Parallel computing with dask +Parallel computing with Dask ============================ -xarray integrates with `dask `__ to support parallel +xarray integrates with `Dask `__ to support parallel computations and streaming computation on datasets that don't fit into memory. -Currently, dask is an entirely optional feature for xarray. However, the -benefits of using dask are sufficiently strong that dask may become a required +Currently, Dask is an entirely optional feature for xarray. However, the +benefits of using Dask are sufficiently strong that Dask may become a required dependency in a future version of xarray. -For a full example of how to use xarray's dask integration, read the -`blog post introducing xarray and dask`_. +For a full example of how to use xarray's Dask integration, read the +`blog post introducing xarray and Dask`_. -.. _blog post introducing xarray and dask: http://stephanhoyer.com/2015/06/11/xray-dask-out-of-core-labeled-arrays/ +.. _blog post introducing xarray and Dask: http://stephanhoyer.com/2015/06/11/xray-dask-out-of-core-labeled-arrays/ -What is a dask array? +What is a Dask array? --------------------- .. image:: _static/dask_array.png :width: 40 % :align: right - :alt: A dask array + :alt: A Dask array Dask divides arrays into many small pieces, called *chunks*, each of which is presumed to be small enough to fit into memory. -Unlike NumPy, which has eager evaluation, operations on dask arrays are lazy. +Unlike NumPy, which has eager evaluation, operations on Dask arrays are lazy. Operations queue up a series of tasks mapped over blocks, and no computation is performed until you actually ask values to be computed (e.g., to print results to your screen or write to disk). At that point, data is loaded into memory and computation proceeds in a streaming fashion, block-by-block. The actual computation is controlled by a multi-processing or thread pool, -which allows dask to take full advantage of multiple processors available on +which allows Dask to take full advantage of multiple processors available on most modern computers. -For more details on dask, read `its documentation `__. +For more details on Dask, read `its documentation `__. .. _dask.io: Reading and writing data ------------------------ -The usual way to create a dataset filled with dask arrays is to load the +The usual way to create a dataset filled with Dask arrays is to load the data from a netCDF file or files. You can do this by supplying a ``chunks`` argument to :py:func:`~xarray.open_dataset` or using the :py:func:`~xarray.open_mfdataset` function. @@ -69,10 +69,10 @@ argument to :py:func:`~xarray.open_dataset` or using the ds = xr.open_dataset('example-data.nc', chunks={'time': 10}) ds -In this example ``latitude`` and ``longitude`` do not appear in the -``chunks`` dict, so only one chunk will be used along those dimensions. It -is also entirely equivalent to open a dataset using ``open_dataset`` and -then chunk the data use the ``chunk`` method, e.g., +In this example ``latitude`` and ``longitude`` do not appear in the ``chunks`` +dict, so only one chunk will be used along those dimensions. It is also +entirely equivalent to opening a dataset using ``open_dataset`` and then +chunking the data using the ``chunk`` method, e.g., ``xr.open_dataset('example-data.nc').chunk({'time': 10})``. To open multiple files simultaneously, use :py:func:`~xarray.open_mfdataset`:: @@ -82,21 +82,21 @@ To open multiple files simultaneously, use :py:func:`~xarray.open_mfdataset`:: This function will automatically concatenate and merge dataset into one in the simple cases that it understands (see :py:func:`~xarray.auto_combine` for the full disclaimer). By default, ``open_mfdataset`` will chunk each -netCDF file into a single dask array; again, supply the ``chunks`` argument to -control the size of the resulting dask arrays. In more complex cases, you can +netCDF file into a single Dask array; again, supply the ``chunks`` argument to +control the size of the resulting Dask arrays. In more complex cases, you can open each file individually using ``open_dataset`` and merge the result, as described in :ref:`combining data`. You'll notice that printing a dataset still shows a preview of array values, -even if they are actually dask arrays. We can do this quickly -with dask because we only need to the compute the first few values (typically -from the first block). To reveal the true nature of an array, print a DataArray: +even if they are actually Dask arrays. We can do this quickly with Dask because +we only need to compute the first few values (typically from the first block). +To reveal the true nature of an array, print a DataArray: .. ipython:: python ds.temperature -Once you've manipulated a dask array, you can still write a dataset too big to +Once you've manipulated a Dask array, you can still write a dataset too big to fit into memory back to disk by using :py:meth:`~xarray.Dataset.to_netcdf` in the usual way. @@ -105,7 +105,7 @@ usual way. ds.to_netcdf('manipulated-example-data.nc') By setting the ``compute`` argument to ``False``, :py:meth:`~xarray.Dataset.to_netcdf` -will return a dask delayed object that can be computed later. +will return a Dask delayed object that can be computed later. .. ipython:: python @@ -117,43 +117,43 @@ will return a dask delayed object that can be computed later. .. note:: - When using dask's distributed scheduler to write NETCDF4 files, + When using Dask's distributed scheduler to write NETCDF4 files, it may be necessary to set the environment variable `HDF5_USE_FILE_LOCKING=FALSE` to avoid competing locks within the HDF5 SWMR file locking scheme. Note that - writing netCDF files with dask's distributed scheduler is only supported for + writing netCDF files with Dask's distributed scheduler is only supported for the `netcdf4` backend. -A dataset can also be converted to a dask DataFrame using :py:meth:`~xarray.Dataset.to_dask_dataframe`. +A dataset can also be converted to a Dask DataFrame using :py:meth:`~xarray.Dataset.to_dask_dataframe`. .. ipython:: python df = ds.to_dask_dataframe() df -Dask DataFrames do not support multi-indexes so the coordinate variables from the dataset are included as columns in the dask DataFrame. +Dask DataFrames do not support multi-indexes so the coordinate variables from the dataset are included as columns in the Dask DataFrame. -Using dask with xarray +Using Dask with xarray ---------------------- Nearly all existing xarray methods (including those for indexing, computation, concatenating and grouped operations) have been extended to work automatically -with dask arrays. When you load data as a dask array in an xarray data -structure, almost all xarray operations will keep it as a dask array; when this +with Dask arrays. When you load data as a Dask array in an xarray data +structure, almost all xarray operations will keep it as a Dask array; when this is not possible, they will raise an exception rather than unexpectedly loading -data into memory. Converting a dask array into memory generally requires an -explicit conversion step. One noteable exception is indexing operations: to +data into memory. Converting a Dask array into memory generally requires an +explicit conversion step. One notable exception is indexing operations: to enable label based indexing, xarray will automatically load coordinate labels into memory. -The easiest way to convert an xarray data structure from lazy dask arrays into -eager, in-memory numpy arrays is to use the :py:meth:`~xarray.Dataset.load` method: +The easiest way to convert an xarray data structure from lazy Dask arrays into +eager, in-memory NumPy arrays is to use the :py:meth:`~xarray.Dataset.load` method: .. ipython:: python ds.load() You can also access :py:attr:`~xarray.DataArray.values`, which will always be a -numpy array: +NumPy array: .. ipython:: :verbatim: @@ -177,7 +177,7 @@ Explicit conversion by wrapping a DataArray with ``np.asarray`` also works: ... Alternatively you can load the data into memory but keep the arrays as -dask arrays using the :py:meth:`~xarray.Dataset.persist` method: +Dask arrays using the :py:meth:`~xarray.Dataset.persist` method: .. ipython:: @@ -191,8 +191,8 @@ sure that your dataset is not larger than available memory. For performance you may wish to consider chunk sizes. The correct choice of chunk size depends both on your data and on the operations you want to perform. -With xarray, both converting data to a dask arrays and converting the chunk -sizes of dask arrays is done with the :py:meth:`~xarray.Dataset.chunk` method: +With xarray, both converting data to a Dask arrays and converting the chunk +sizes of Dask arrays is done with the :py:meth:`~xarray.Dataset.chunk` method: .. ipython:: python :suppress: @@ -216,7 +216,7 @@ along a particular dimension, an exception is raised when you try to access .. note:: - In the future, we would like to enable automatic alignment of dask + In the future, we would like to enable automatic alignment of Dask chunksizes (but not the other way around). We might also require that all arrays in a dataset share the same chunking alignment. Neither of these are currently done. @@ -224,17 +224,17 @@ along a particular dimension, an exception is raised when you try to access NumPy ufuncs like ``np.sin`` currently only work on eagerly evaluated arrays (this will change with the next major NumPy release). We have provided replacements that also work on all xarray objects, including those that store -lazy dask arrays, in the :ref:`xarray.ufuncs ` module: +lazy Dask arrays, in the :ref:`xarray.ufuncs ` module: .. ipython:: python import xarray.ufuncs as xu xu.sin(rechunked) -To access dask arrays directly, use the new +To access Dask arrays directly, use the new :py:attr:`DataArray.data ` attribute. This attribute exposes -array data either as a dask array or as a numpy array, depending on whether it has been -loaded into dask or not: +array data either as a Dask array or as a NumPy array, depending on whether it has been +loaded into Dask or not: .. ipython:: python @@ -243,25 +243,26 @@ loaded into dask or not: .. note:: In the future, we may extend ``.data`` to support other "computable" array - backends beyond dask and numpy (e.g., to support sparse arrays). + backends beyond Dask and NumPy (e.g., to support sparse arrays). .. _dask.automatic-parallelization: Automatic parallelization ------------------------- -Almost all of xarray's built-in operations work on dask arrays. If you want to -use a function that isn't wrapped by xarray, one option is to extract dask -arrays from xarray objects (``.data``) and use dask directly. +Almost all of xarray's built-in operations work on Dask arrays. If you want to +use a function that isn't wrapped by xarray, one option is to extract Dask +arrays from xarray objects (``.data``) and use Dask directly. Another option is to use xarray's :py:func:`~xarray.apply_ufunc`, which can -automate `embarrassingly parallel `__ -"map" type operations where a functions written for processing NumPy arrays -should be repeatedly applied to xarray objects containing dask arrays. It works -similarly to :py:func:`dask.array.map_blocks` and :py:func:`dask.array.atop`, -but without requiring an intermediate layer of abstraction. - -For the best performance when using dask's multi-threaded scheduler, wrap a +automate `embarrassingly parallel +`__ "map" type operations +where a function written for processing NumPy arrays should be repeatedly +applied to xarray objects containing Dask arrays. It works similarly to +:py:func:`dask.array.map_blocks` and :py:func:`dask.array.atop`, but without +requiring an intermediate layer of abstraction. + +For the best performance when using Dask's multi-threaded scheduler, wrap a function that already releases the global interpreter lock, which fortunately already includes most NumPy and Scipy functions. Here we show an example using NumPy operations and a fast function from @@ -311,7 +312,7 @@ work as a streaming operation, when run on arrays loaded from disk: In [58]: array2 = array1 + 0.5 * rs.randn(1000, 100000) - # using one core, on numpy arrays + # using one core, on NumPy arrays In [61]: %time _ = spearman_correlation(array1, array2, 'time') CPU times: user 21.6 s, sys: 2.84 s, total: 24.5 s Wall time: 24.9 s @@ -320,7 +321,7 @@ work as a streaming operation, when run on arrays loaded from disk: In [9]: chunked2 = array2.chunk({'place': 10}) - # using all my laptop's cores, with dask + # using all my laptop's cores, with Dask In [63]: r = spearman_correlation(chunked1, chunked2, 'time').compute() In [64]: %time _ = r.compute() @@ -336,17 +337,17 @@ multiple chunks along a core dimension: In [63]: spearman_correlation(chunked1, chunked2, 'place') ValueError: dimension 'place' on 0th function argument to apply_ufunc with dask='parallelized' consists of multiple chunks, but is also a core - dimension. To fix, rechunk into a single dask array chunk along this + dimension. To fix, rechunk into a single Dask array chunk along this dimension, i.e., ``.rechunk({'place': -1})``, but beware that this may significantly increase memory usage. -The reflects the nature of core dimensions, in contrast to broadcast -(non-core) dimensions that allow operations to be split into arbitrary chunks -for application. +This reflects the nature of core dimensions, in contrast to broadcast (non-core) +dimensions that allow operations to be split into arbitrary chunks for +application. .. tip:: - For the majority of NumPy functions that are already wrapped by dask, it's + For the majority of NumPy functions that are already wrapped by Dask, it's usually a better idea to use the pre-existing ``dask.array`` function, by using either a pre-existing xarray methods or :py:func:`~xarray.apply_ufunc()` with ``dask='allowed'``. Dask can often @@ -357,19 +358,19 @@ for application. Chunking and performance ------------------------ -The ``chunks`` parameter has critical performance implications when using dask +The ``chunks`` parameter has critical performance implications when using Dask arrays. If your chunks are too small, queueing up operations will be extremely -slow, because dask will translates each operation into a huge number of -operations mapped across chunks. Computation on dask arrays with small chunks -can also be slow, because each operation on a chunk has some fixed overhead -from the Python interpreter and the dask task executor. +slow, because Dask will translate each operation into a huge number of +operations mapped across chunks. Computation on Dask arrays with small chunks +can also be slow, because each operation on a chunk has some fixed overhead from +the Python interpreter and the Dask task executor. Conversely, if your chunks are too big, some of your computation may be wasted, -because dask only computes results one chunk at a time. +because Dask only computes results one chunk at a time. -A good rule of thumb to create arrays with a minimum chunksize of at least one -million elements (e.g., a 1000x1000 matrix). With large arrays (10+ GB), the -cost of queueing up dask operations can be noticeable, and you may need even +A good rule of thumb is to create arrays with a minimum chunksize of at least +one million elements (e.g., a 1000x1000 matrix). With large arrays (10+ GB), the +cost of queueing up Dask operations can be noticeable, and you may need even larger chunksizes. .. ipython:: python @@ -381,10 +382,10 @@ larger chunksizes. Optimization Tips ----------------- -With analysis pipelines involving both spatial subsetting and temporal resampling, dask performance can become very slow in certain cases. Here are some optimization tips we have found through experience: +With analysis pipelines involving both spatial subsetting and temporal resampling, Dask performance can become very slow in certain cases. Here are some optimization tips we have found through experience: -1. Do your spatial and temporal indexing (e.g. ``.sel()`` or ``.isel()``) early in the pipeline, especially before calling ``resample()`` or ``groupby()``. Grouping and rasampling triggers some computation on all the blocks, which in theory should commute with indexing, but this optimization hasn't been implemented in dask yet. (See `dask issue #746 `_). +1. Do your spatial and temporal indexing (e.g. ``.sel()`` or ``.isel()``) early in the pipeline, especially before calling ``resample()`` or ``groupby()``. Grouping and resampling triggers some computation on all the blocks, which in theory should commute with indexing, but this optimization hasn't been implemented in Dask yet. (See `Dask issue #746 `_). -2. Save intermediate results to disk as a netCDF files (using ``to_netcdf()``) and then load them again with ``open_dataset()`` for further computations. For example, if subtracting temporal mean from a dataset, save the temporal mean to disk before subtracting. Again, in theory, dask should be able to do the computation in a streaming fashion, but in practice this is a fail case for the dask scheduler, because it tries to keep every chunk of an array that it computes in memory. (See `dask issue #874 `_) +2. Save intermediate results to disk as a netCDF files (using ``to_netcdf()``) and then load them again with ``open_dataset()`` for further computations. For example, if subtracting temporal mean from a dataset, save the temporal mean to disk before subtracting. Again, in theory, Dask should be able to do the computation in a streaming fashion, but in practice this is a fail case for the Dask scheduler, because it tries to keep every chunk of an array that it computes in memory. (See `Dask issue #874 `_) 3. Specify smaller chunks across space when using ``open_mfdataset()`` (e.g., ``chunks={'latitude': 10, 'longitude': 10}``). This makes spatial subsetting easier, because there's no risk you will load chunks of data referring to different chunks (probably not necessary if you follow suggestion 1).