issue_comments: 656403217

Here's an alternative map_blocks solution:

``` python def write_block(ds, t0): if len(ds.time) > 0: fname = (ds.time[0] - t0).values.astype("timedelta64[h]").astype(int) ds.to_netcdf(f"temp/file-{fname:06d}.nc")

# dummy return
return ds.time

ds = xr.tutorial.open_dataset("air_temperature", chunks={"time": 100}) ds.map_blocks(write_block, kwargs=dict(t0=ds.time[0])).compute(scheduler="processes") ```

There are two workarounds here though. 1. The user function always has to return something. 2. We can't provide template=ds.time because it has no chunk information and ds.time.chunk({"time": 100}) silently does nothing because it is an IndexVariable. So the user function still needs the len(ds.time) > 0 workaround.

I think a cleaner API may be to have dask.compute([write_block(block) for block in ds.to_delayed()]) where ds.to_delayed() yields a bunch of tasks; each of which gives a Dataset wrapping one block of the underlying array.