Join or view over the same chunk across multiple sub-groups in DirectoryStore

[sanjaysrikakulam]

Hi,

Is it possible to create a view or join multiple 1D arrays stored as a separate sub-group in a DirectoryStore?

Scenario:
I generate 200K bloom filters (1D bool NumPy array for example) and store each bloom filter as a separate sub-group (under the main group/dataset named BloomFilters) and each of these bloom filters is stored as a single column in each sub-group. So is it possible to create a view across a chunk (for example, the row chunk size is a million) of all sub-groups?

Note: Each sub-group or bloom filter will be created in parallel and will be written as a sub-group under the main group/dataset in parallel and all of the sub-groups will have the same dimensions.

[joshmoore]

Not that I know of. I assume it would need to be implemented as its own Store.

[sanjaysrikakulam]

OK, thank you for the information.

[jakirkham]

Perhaps in the way that this is consumed by another library (like Dask), there could be a way to handle this. That said, I don't fully understand what is being tried here (an example would likely help)

[sanjaysrikakulam]

Hi @jakirkham

The problem I am trying to solve is how to efficiently (reduced memory, on-disk size and performance) store each bloom filter (in this case 200K for example) of size 10B bits/bool so that they can later be easily accessed row-wise. Since I cannot build a NumPy matrix of that size in memory, I was planning to save them as individual sub-group in Zarr in parallel and then if there was a view or join like feature it will help me to slice n rows easily instead of looping through every sub-group and then reading n rows from each and building a small NumPy array. I construct this NumPy array to get non-zero indices in every row which I later use for fast querying. This is more of a membership testing and indexing.

Some code snippets:

import os
import zarr
from numcodecs import Blosc, PackBits
import numpy as np
from multiprocessing import Process

# Zarr store related #
store_file = "data/bloomfilters.zarr"
store = zarr.DirectoryStore(store_file)
sync_dir = os.path.splitext(store_file)[0] + ".sync"
synchronizer = zarr.ProcessSynchronizer(sync_dir)
Blosc.use_threads = False
compressor = Blosc(cname="zstd", clevel=9, shuffle=Blosc.BITSHUFFLE)
root_group = zarr.group(store=store, synchronizer=synchronizer)
bf_group = root_group.create_group(name="BloomFilterIndex")

# Parallelize bloom filter writes to zarr #
bloomfilter_size = 10000000000
n_cores = 64
processes = []
for samp_list in np.array_split(samples_list, n_cores):
    proc = Process(target=write_bf_to_zarr, args=(bf_group, compressor, bloomfilter_size, samples_list))
    proc.start()
    processes.append(proc)

for proc in processes:
    proc.join()

# Write function #
def write_bf_to_zarr(bf_group_handle, compressor, bloomfilter_size, samples_list):
    for sample_info in samples_list:
        col_idx = sample_info[0]
        file_paths = sample_info[1]
        bloomfilter = bf_group_handle.zeros(
            f"sample_{col_idx}",
            shape=(bloomfilter_size),
            chunks=(int(bloomfilter_size/8)),
            dtype="bool",
            filters=[PackBits()],
            compressor=compressor,
        )
        # Note: bloomfilter_cython will create bloom filter as NumPy boolean array and will return it
        bloomfilter[:] = bloomfilter_cython(file_paths, bloomfilter_size)

# To slice 'n' rows (example 10 rows) from each of the sub-group
m = len(bf_group.keys())
n = 10
n_rows = np.zeros((n, m), dtype=bool)

start = 0
stop = 10
for col_idx, key in enumerate(bf_group.keys()):
    n_rows[:, col_idx] = bf_group.get(key)[start:stop]   

So instead of reading from every sub-group in a for loop, I was wondering if there is like a view or join (as in joining database tables) to easily extract n rows which I then could simply use to extract non-zero indices of every row.

Please suggest if you have any ideas to improve or do this in a better way.

Thank you!

[sanjaysrikakulam]

Is extracting 'n' rows from each sub group via dask possible? Any example or help there would be really great as i am totally new to dask and could not find any example resembling to what i am trying to do.