# Xarray to Zarr Parallel Writes with Dask Distributed

**URL:** <https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611>\
**Category:** Data\
**Created:** [July 25, 2022, 8:07pm UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611 "2022-07-25T20:07:41Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![jgreen](https://yyz2.discourse-cdn.com/flex030/user_avatar/discourse.pangeo.io/jgreen/32/1683_2.png) [@jgreen](https://discourse.pangeo.io/u/jgreen)\
**Post date:** [July 25, 2022, 8:07pm UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611/1 "2022-07-25T20:07:41Z")

</div>

Hello, while messing with converting NetCDF-4 datasets to Zarr using Xarray, a sudden point of confusion struck me:

I know from reading the Xarray documentation that lazily representing the data as a Dask array allows for easy configuration of chunk sizes, and the writes do not take up vast amounts of memory. This leads me to believe that they are done out of core by default, even when Dask Distributed is not set up as a local cluster on a single machine. I am a little confused with the inner-workings of this conversion.

Without setting up a local cluster, does Dask automatically parallelize the write process when `xarray.dataset.to_zarr(...)` is called? Would the amount of workers used for the write correspond to the number of computing resources available in the machine (i.e. a single worker per vCPU in a VM)?

For example, what is the difference in what is happening internally between using Dask Distributed:

```auto
from dask.distributed import Client, LocalCluster
cluster = LocalCluster(threads_per_worker=1)
client = Client(cluster)

import xarray as xr
import dask
import intake
da = intake.open_netcdf(<URI>).to_dask().<DATA VARIABLE>
da = da.chunk(chunks = dict(zip(<COORDINATES>, <CHUNKSIZE>)))
ds = da.to_dataset()

ds.to_zarr(store=<NEW URI>, consolidated=True)

```

and just Xarray & a Dask array:

```auto
import xarray as xr
import dask
import intake
da = intake.open_netcdf(<URI>).to_dask().<DATA VARIABLE>
da = da.chunk(chunks = dict(zip(<COORDINATES>, <CHUNKSIZE>)))
ds = da.to_dataset()

ds.to_zarr(store=<NEW URI>, consolidated=True)

```

---

<div class="post-metadata">

**Author:** ![raybellwaves](https://yyz2.discourse-cdn.com/flex030/user_avatar/discourse.pangeo.io/raybellwaves/32/782_2.png) [@raybellwaves](https://discourse.pangeo.io/u/raybellwaves)\
**Post date:** [July 25, 2022, 11:14pm UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611/2 "2022-07-25T23:14:21Z")

</div>

Mind sending screenshots using viztracer. I find it’s a good tool for seeing all the func calls on a line/script see [to\_zarr to s3 with asynchronous=False · Discussion #5869 · pydata/xarray · GitHub](https://github.com/pydata/xarray/discussions/5869)

---

<div class="post-metadata">

**Author:** ![willirath](https://yyz2.discourse-cdn.com/flex030/user_avatar/discourse.pangeo.io/willirath/32/122_2.png) [@willirath](https://discourse.pangeo.io/u/willirath)\
**Post date:** [July 26, 2022, 7:52am UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611/3 "2022-07-26T07:52:31Z")

</div>

Hi @jgreen,

your assumption is correct: Dask will also work without _explicitly_ specifying a cluster. If you don’t set up a cluster yourself, dask will spin one up (and just assume that the whole machine is yours) [from here](https://github.com/dask/distributed/blob/55cc1a50f6fc7c10815bed93681eba970c9b63f0/distributed/deploy/local.py#L23) and [use a heuristic](https://github.com/dask/distributed/blob/55cc1a50f6fc7c10815bed93681eba970c9b63f0/distributed/deploy/utils.py#L9) to determine the number of processes and number of threads per cluster.

As [the heuristic](https://github.com/dask/distributed/blob/55cc1a50f6fc7c10815bed93681eba970c9b63f0/distributed/deploy/utils.py#L9) won’t necessarily result in `threads_per_worker=1`, your two snippets likely end up working with a differently shaped cluster.

---

<div class="post-metadata">

**Author:** ![willirath](https://yyz2.discourse-cdn.com/flex030/user_avatar/discourse.pangeo.io/willirath/32/122_2.png) [@willirath](https://discourse.pangeo.io/u/willirath)\
**Post date:** [July 26, 2022, 8:03am UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611/4 "2022-07-26T08:03:10Z")

</div>

A few remarks on

```python
da = intake.open_netcdf(<URI>).to_dask().<DATA VARIABLE>
da = da.chunk(chunks = dict(zip(<COORDINATES>, <CHUNKSIZE>)))
ds = da.to_dataset()

ds.to_zarr(store=<NEW URI>, consolidated=True)

```

If I’m not mistaken, `intake.open_netcdf()` is basically an interface to `xarray.open_mfdataset()`. This should result in `da` containing a Dask array right away and in that sense is out of core directly.

However, as there is no chunking specified, `intake.open_netcdf(<URI>).to_dask().<DATA VARIABLE>` will likely consist of one large chunk spanning all of the array. The subsequent `da.chunk()` call will add some re-chunking tasks to this and hence the final version of `da` or `ds` will have the desired chunking.

But the _read_ operation on the netCDF file will always get the huge full-array chunk which was implicitly requested in the initial opening statement. This can have huge performance implications and is probably not what you had in mind. (You’re not alone with this. The `open().chunk().do_something().store()` paradigm is feeling pretty natural to most people and I’ve seen it hundreds of times.)

To really work out of core on chunks smaller than the full array, you’d need to specify chunks right at the opening step. I’m not too familiar with `intake.open_netcdf()`, but for the underlying Xarray method, the following should work the way you likely intended this to behave:

```python
da = xarray.open_dataset(
    <URI>,
    chunks=dict(zip(<COORDINATES>, <CHUNKSIZE>)),
).<DATA VARIABLE>

```

---

<div class="post-metadata">

**Author:** ![willirath](https://yyz2.discourse-cdn.com/flex030/user_avatar/discourse.pangeo.io/willirath/32/122_2.png) [@willirath](https://discourse.pangeo.io/u/willirath)\
**Post date:** [July 26, 2022, 8:12am UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611/5 "2022-07-26T08:12:15Z")

</div>

And then there’s the internal serialisation of multi-threaded netCDF reads…

Running

```python
print(
    xr.open_dataset(
        <FILE-W-10-TIMESTEPS>,
        chunks={"time": 1},
    )
    .max()
    .compute()
)

```

with

```python
multithreaded_only_cluster = dask.distributed.LocalCluster(
    n_workers=1,
    threads_per_worker=10,
)

```

will have 10 tasks for reading the 10 chuks we split the array into in time direction, but these tasks will all wait for each other because netCDF / HDF5 will internally serialise them.

Running it with

```python
multiprocs_only_cluster = dask.distributed.LocalCluster(
    n_workers=10,
    threads_per_worker=1,
)

```

however, will (on most systems) actually run 10 parallel read ops on the file.

As Dask’s memory management is more stable in the multithreaded world, a mixture of threads and processes is typically the way to go (and hence implemented in the heuristics I’ve linked above).

---

<div class="post-metadata">

**Author:** ![jgreen](https://yyz2.discourse-cdn.com/flex030/user_avatar/discourse.pangeo.io/jgreen/32/1683_2.png) [@jgreen](https://discourse.pangeo.io/u/jgreen)\
**Post date:** [July 26, 2022, 2:14pm UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611/6 "2022-07-26T14:14:51Z")

</div>

Thanks @willirath, I appreciate the information! The reason I am using `intake.open_netcdf` is because I found it very difficult to get the `xr.open_dataset(...)` engines to work with Google Cloud object storage. The `intake` module has proven much easier in this regard.

I have the NetCDF4 chunks internally stored at 100MB apiece, and from a [previous post](https://discourse.pangeo.io/t/unclear-behavior-of-netcdf4-files-loaded-with-intake-xarray/2529/7) I know that the Dask array chunks must be configured to exactly match – or be exact multiples – of the NetCDF internal chunks for an efficient parallel read. So, the chunksize I am specifying for the Dask array comes from `intake.open_netcdf(<URI>).to_dask().<DATA VARIABLE>.encoding['chunksizes']`

I will try the write using the mixture of threads/processes and implement the heuristic, I’m eager to see what new information I can gather from this. You really cleared things up for me!

---

<div class="post-metadata">

**Author:** ![jgreen](https://yyz2.discourse-cdn.com/flex030/user_avatar/discourse.pangeo.io/jgreen/32/1683_2.png) [@jgreen](https://discourse.pangeo.io/u/jgreen)\
**Post date:** [July 26, 2022, 2:17pm UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611/7 "2022-07-26T14:17:16Z")

</div>

@raybellwaves I think I will start to use this tool more often, it looks very helpful in determining what’s going on under the hood. Thanks for pointing me to it!

---

<div class="post-metadata">

**Author:** ![martindurant](https://yyz2.discourse-cdn.com/flex030/user_avatar/discourse.pangeo.io/martindurant/32/388_2.png) [@martindurant](https://discourse.pangeo.io/u/martindurant)\
**Post date:** [July 26, 2022, 2:51pm UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611/8 "2022-07-26T14:51:18Z")

</div>

Couple of clarifications:

- when you call `to_dask` on the intake source, you are effectively specifying `chunks={}` to xarray, which means it will use whatever is the native chunksize and is often what you want. The repr of the resulting dataset will tell you the chunks the underlying dask arrays are using.
- using dask from xarray will _not_ create a LocalCluster for you, but use an in-process thread pool. That might be fine, but adding `client = dask.distributed.Client(...)` lets you create a cluster explicitly and control for all of the configurable options.

You may wish to check out pangeo-forge, which provides a mechanism for executing netCDF-\>zarr workflows from prescriptive recipes.

Also, you might be interested in the kerchunk project, which can give you a zarr-based view on your original netCDF files without a transform&copy, and provide scalable parallel access that way.

---

<div class="post-metadata">

**Author:** ![jgreen](https://yyz2.discourse-cdn.com/flex030/user_avatar/discourse.pangeo.io/jgreen/32/1683_2.png) [@jgreen](https://discourse.pangeo.io/u/jgreen)\
**Post date:** [July 26, 2022, 3:32pm UTC](https://discourse.pangeo.io/t/xarray-to-zarr-parallel-writes-with-dask-distributed/2611/9 "2022-07-26T15:32:13Z")

</div>

> when you call `to_dask` on the intake source, you are effectively specifying `chunks={}` to xarray, which means it will use whatever is the native chunksize and is often what you want.

I should have clarified, I was mainly using the chunksize option to write to different Zarr chunksizes for some benchmarking I was doing. Though, since I have 3 different NetCDF files with the desired chunking schemes, it seems that I was just adding an unnecessary line of code.

Defining the local cluster sped up my writes by a considerable amount, so I think going forward with that approach may be the best method in this context.