Writing to Zarr from Dask

[jmdelahanty]

Hello fellow Zarr enthusiasts!

I'm currently trying to write a dask.array into Zarr using the dask.array.to_zarr() functionality in this project.

It seems that when I try to write to Zarr with this command:

with dask.config.set(scheduler='processes'):
    da.to_zarr(hog_images, "/scratch/snlkt_facial_expression/hog_images/data.zarr", compressor=compressor)

I get the errors below:

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/multiprocessing/spawn.py", line 116, in spawn_main
    exitcode = _main(fd, parent_sentinel)
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/multiprocessing/spawn.py", line 125, in _main
    prepare(preparation_data)
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/multiprocessing/spawn.py", line 236, in prepare
    _fixup_main_from_path(data['init_main_from_path'])
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/multiprocessing/spawn.py", line 287, in _fixup_main_from_path
    main_content = runpy.run_path(main_path,
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/runpy.py", line 265, in run_path
    return _run_module_code(code, init_globals, run_name,
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/runpy.py", line 97, in _run_module_code
    _run_code(code, mod_globals, init_globals,
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/runpy.py", line 87, in _run_code
    exec(code, run_globals)
  File "/nadata/snlkt/data/facial_expression/specialk/dask_faces.py", line 633, in <module>
    da.to_zarr(hog_images, "/scratch/snlkt_facial_expression/hog_images/data.zarr", compressor=compressor)
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/dask/array/core.py", line 3503, in to_zarr
    z = zarr.create(
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/zarr/creation.py", line 149, in create
    init_array(store, shape=shape, chunks=chunks, dtype=dtype, compressor=compressor,
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/zarr/storage.py", line 350, in init_array
    _init_array_metadata(store, shape=shape, chunks=chunks, dtype=dtype,
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/zarr/storage.py", line 381, in _init_array_metadata
    raise ContainsArrayError(path)
zarr.errors.ContainsArrayError: path '' contains an array
Traceback (most recent call last):
  File "dask_faces.py", line 633, in <module>
    da.to_zarr(hog_images, "/scratch/snlkt_facial_expression/hog_images/data.zarr", compressor=compressor)
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/dask/array/core.py", line 3512, in to_zarr
    return arr.store(z, lock=False, compute=compute, return_stored=return_stored)
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/dask/array/core.py", line 1689, in store
    r = store([self], [target], **kwargs)
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/dask/array/core.py", line 1163, in store
    compute_as_if_collection(Array, store_dsk, map_keys, **kwargs)
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/dask/base.py", line 317, in compute_as_if_collection
    return schedule(dsk2, keys, **kwargs)
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/dask/multiprocessing.py", line 220, in get
    result = get_async(
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/site-packages/dask/local.py", line 495, in get_async
    for key, res_info, failed in queue_get(queue).result():
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/concurrent/futures/_base.py", line 437, in result
    return self.__get_result()
  File "/home/jdelahanty/miniconda3/envs/facial_expression/lib/python3.8/concurrent/futures/_base.py", line 389, in __get_result
    raise self._exception
concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.

So I'm guessing that what's happening is my processes are ending because of the Zarr exception where it can't write to disk. It seems like the first chunk is written to disk properly but then when the next chunk is getting written it states that there's already an array!

I know there's the append mode for zarr, but I can't seem to supply that to to_zarr method. I'm not even sure that's the issue in the first place!

So any advice for writing to Zarr from Dask?

[jmdelahanty]

Perhaps related to this issue?

It turns out that this does indeed work if only I had actually written my code properly! I wrote here that I was using it correctly, as in:

to_zarr(array, "path/to/data.zarr")

But in reality, my script had things written like this:

to_zarr("path/to/data.zarr", array)

But these other issues are really interesting. I don't know if I can help solve them, but maybe I can try!

[jmdelahanty]

So if you give a dask array to zarr it's like a non-parallel-aware layer is trying to "manage" a parallel-aware one, not great.

I see, that makes more sense now I think. Basically they don't really know how to talk to each other/ensure they play nice.

EDIT
The code that this is running on did complete (eventually) using default settings which I understand is the threaded scheduler?

I wonder if this is a matter of the chunk sizes not matching up.

That's possible, I had thought that we had properly used map_blocks to get the outputs of hog (a HOG image and it's descriptors) into their own separate dask arrays, maybe there's something going wrong in that part of the code.

The only other chunk size difference that I can think of is that the number of frames read into dask_image.imread.imread doesn't divide nicely into the number of frames. But I think Zarr and Dask don't really mind that right?

[jmdelahanty]

Another update here: @ParticularMiner got this to run on their laptop which was running an upgraded version of Dask. I made a new environment with an upgraded Dask/Zarr but am still running into the same error.

The only difference now appears to be that they were running this on Windows while I'm trying to run it on a Linux machine.

Could that be the difference somehow @joshmoore ?

[joshmoore]

Perhaps, e.g., if Windows did some locking itself?

[jmdelahanty]

Is there a way that I can tell or profile that? Or a way to tell my Linux computer to perform locking? I was checking out the dask.delayed docs as a potential solution, but it’s a bit over my head.

Perhaps it’s the difference between spawn and fork? Just looking here. Not sure how to use Spawn over Fork though…

[jmdelahanty]

It turns out that this does indeed work if only I had actually written my code properly! I wrote here that I was using it correctly, as in:

to_zarr(array, "path/to/data.zarr")

But in reality, my script had things written like this:

to_zarr("path/to/data.zarr", array)

But these other issues are really interesting. I don't know if I can help solve them, but maybe I can try!