Local Data Processing Store¶

Local filesystem implementation of the data processing store.

The store exposes a local directory as a dataset storage root. Relative locations are resolved below that root, while absolute paths are passed through unchanged. It provides the file and parquet operations used by the processing pipeline: directory management, source-file discovery, local file access, lazy parquet scans, bounded-memory parquet iteration, one-shot table writes, and chunked table writers.

The root must be absolute. Passing create=True creates the root directory before validation, which is useful for scratch stores and test fixtures.

Examples:

Discover raw Excel files while using the ingest spec to ignore excluded files such as README.xlsx:

>>> from pathlib import Path
>>> import polars as pl
>>> from batgrad.contracts.mapping import DatasetStageId
>>> from batgrad.data.datasets.pozzato_2022.config import DATASET_SPEC
>>> from batgrad.storage.local import LocalDataProcessingStore
>>> store = LocalDataProcessingStore(Path("/data/loc_datasets"))
>>> raw_spec = DATASET_SPEC.processing_stages[DatasetStageId.ingested]
>>> raw_root = DATASET_SPEC.source_root(DatasetStageId.raw)
>>> paths = []
>>> for pattern in raw_spec.included_file_patterns:
...     paths.extend(
...         path
...         for path in store.list_files(raw_root, pattern=pattern)
...         if raw_spec.is_included_file(path)
...     )

Read a parquet table lazily when the whole result does not need to be loaded immediately:

>>> frame = store.scan_table(
...     "normalized/cell.parquet",
...     columns=("voltage",),
...     filters=pl.col("voltage") > 3.6,
... ).collect()

Iterate over a full table in bounded-size chunks:

>>> def process(chunk):
...     pass
>>> for chunk in store.iter_table_chunks("normalized/cell.parquet", 100_000):
...     process(chunk)

Extract explicit row windows when a manifest points to table slices:

>>> slices = ((0, 1024), (10_000, 2048))
>>> for chunk in store.iter_table_slices("normalized/cell.parquet", slices, 512):
...     process(chunk)

Parameters:

• root (str | Path) –

  Absolute filesystem directory used as the store root.

• create (bool, default: False ) –

  Create root before validation if it does not exist.

Raises:

• ValueError –

  If root is not absolute.

• FileNotFoundError –

  If root does not exist and create is false.

• NotADirectoryError –

  If root exists but is not a directory.

153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174

def __init__(self, root: str | Path, *, create: bool = False) -> None:
    """Create a local store rooted at an absolute directory.

    Args:
        root: Absolute filesystem directory used as the store root.
        create: Create `root` before validation if it does not exist.

    Raises:
        ValueError: If `root` is not absolute.
        FileNotFoundError: If `root` does not exist and `create` is false.
        NotADirectoryError: If `root` exists but is not a directory.
    """
    root_path = Path(root)
    if not root_path.is_absolute():
        raise ValueError(f"Local data root must be an absolute path: {root_path}")
    if create:
        root_path.mkdir(parents=True, exist_ok=True)
    if not root_path.exists():
        raise FileNotFoundError(f"Local data root does not exist: {root_path}")
    if not root_path.is_dir():
        raise NotADirectoryError(f"Local data root is not a directory: {root_path}")
    self.root = str(root_path.resolve())