# Choosing Dask chunk sizes for xarray: no OOMs, no idle workers
Dask splits every array into chunks, and one chunk is roughly one task per
operation. Bad chunking fails in three predictable ways:
- **Too small** -- thousands of tiny tasks drown the scheduler in overhead.
- **Too big** -- workers cannot hold several chunks at once, so jobs OOM or spill to disk.
- **Wrong axis** -- reductions and windowed ops shuffle data across chunk boundaries.
## Size rules
- Aim for **100 MB to 1 GB per chunk**; chunks under ~1 MB are almost always bad.
(Dask blog, "Choosing good chunk sizes".)
- Minimum **~1 million elements per chunk** (e.g. 1000x1000); for arrays of 10+ GB
you will need larger chunks. (xarray user guide, "Chunking and performance".)
- Keep the total chunk count below roughly **10,000 to 100,000**. The scheduler
coordinates each task in about 1 ms, so each task should compute for seconds,
not milliseconds.
- A chunk's byte size is `prod(chunk_shape) * dtype.itemsize`.
## Feed all workers
You need **at least as many chunks as worker cores, ideally 2-3x**. Fewer chunks
than cores means cores sit idle; much more means scheduling overhead with no
extra parallelism.
## Memory rule
Chunk sizes must be small enough that **many chunks fit in worker memory at
once** (several blocks per worker when running multi-core). If a single chunk
is a large fraction of a worker's memory, intermediate arrays during reductions
will push it over.
## Chunk along the dimension you reduce
Choose chunk shapes from the downstream analysis, and chunk as early as
possible. Two classic cases:
- **Time-series reductions**: chunk time contiguously, e.g.
`ds.chunk({"time": -1})`, so `ds.mean("time")` reduces inside chunks instead
of across the network.
- **Per-timestep maps**: smaller chunks along time let Dask run the same op on
each time chunk independently.
## Rechunk before operations that span chunks
Rechunking is expensive, so only do it when the layout fights the operation:
- `apply_ufunc(dask="parallelized")` requires a **single chunk along each core
dimension** and raises otherwise; fix with `.rechunk({DIM: -1})` when the full
dimension fits in memory.
- **rolling**: windowed aggregations pull values from neighboring positions, so
chunks smaller than the window force chunk-boundary exchange and can trigger
a rechunk; rechunk the windowed dimension to chunks larger than the window first.
- **groupby / resample**: these trigger computation across all blocks. Subset
with `.sel()` / `.isel()` *before* grouping, and install `flox` for faster
groupby reductions.
## Set chunking early; change it deliberately
```python
# at read time
ds = xr.open_dataset("data.nc", chunks={"time": 10})
ds = xr.open_zarr("store.zarr") # chunked like the on-disk store
ds = xr.open_dataset("data.nc", chunks="auto") # multiples of on-disk chunks
ds = xr.open_dataset("data.nc", chunks={}) # 1 dask chunk per on-disk chunk
# any time
ds = ds.chunk({"time": 100, "lat": 500, "lon": 500})
ds = ds.chunk(time=TimeResampler("MS")) # one chunk per calendar month
```
Notes:
- Dict-style chunks that differ wildly from the on-disk layout are slow and
bandwidth-heavy; match the storage layout when possible.
- After an expensive op like rechunking, `persist()` the result on a cluster
and keep the returned object.
- For large-scale layout changes (e.g. time-chunked simulation data rewritten
to space-chunked), write another copy or use Rechunker rather than in-memory
rechunking.
## Read chunk sizes back from the dask graph
Never guess -- inspect:
```python
a = ds["temp"].data # underlying dask array
a.chunks # tuple-of-tuples: exact block boundaries
a.chunksize # shape of a typical block
a.npartitions # == prod(a.numblocks): number of chunks
a.nbytes / a.npartitions # bytes per chunk (uniform sizes)
```
Compare chunk bytes against the 100 MB-1 GB band, and `a.npartitions` against
your worker-core count. If `dask.array` auto-chunking is active, the
`array.chunk-size` config key (default `"128MiB"`) drives the target.