databricks_unity_parquet_loader

databricks_unity_parquet_loader ¤

DatabricksUnityParquetLoader ¤

DatabricksUnityParquetLoader(
    volume_path: str | None = None,
    *,
    catalog: str | None = None,
    schema: str | None = None,
    volume: str | None = None,
    prefix: str = "",
    base_path: str = "/Volumes",
    hour_pattern: str = "{Y}/{m}/{d}/{H}/",
    validate: bool = True
)

Load canonical parquet files governed by Databricks Unity Catalog.

Unity Catalog exposes the same parquet files that already live in your cloud storage (an external UC Volume over, e.g., an Azure blob container). The on-disk layout is unchanged -- <prefix>/YYYY/MM/DD/HH/<uuid>.parquet with the canonical columns systime, uuid, value_double, value_integer, value_string, value_bool, is_delta -- so the frames returned here flow straight into every ts-shape transformation, event detector, Pipeline and DataIntegratorHybrid without any change.

Designed for use inside Databricks notebooks / pipelines. A UC Volume is FUSE-mounted at /Volumes/<catalog>/<schema>/<volume>/..., so this loader reads parquet directly from that mounted path with pandas.read_parquet. It deliberately keeps the resource footprint low:

No download step and no databricks-sdk / network client -- the mounted Volume is read like a local directory.
Navigates only the hour folders in range (YYYY/MM/DD/HH) instead of scanning the whole tree, so a one-hour query never lists the whole Volume.
Column projection and row-predicate pushdown via columns / filters forwarded to pandas.read_parquet (pyarrow), so only needed bytes are read.
Streaming generators (:meth:stream_by_time_range, :meth:stream_files_by_time_range_and_uuids) yield one frame at a time so the driver never has to hold the full dataset in memory.

Reads are sequential by design (no thread pool) to avoid adding CPU/memory pressure on a shared cluster driver; pushdown + streaming keep them cheap.

Off-cluster (outside Databricks) the same code works against any mounted or synced copy of the Volume; if you cannot mount it, use :class:AzureBlobParquetLoader against the underlying storage instead.

Resolve the mounted Unity Catalog Volume root to read from.

Provide either an explicit volume_path or the catalog / schema / volume triple (joined under base_path)::

# explicit mounted path
DatabricksUnityParquetLoader(
    volume_path="/Volumes/main/plant/timeseries", prefix="parquet",
)

# from catalog/schema/volume parts
DatabricksUnityParquetLoader(
    catalog="main", schema="plant", volume="timeseries",
    prefix="parquet",
)

Parameters:

Name	Type	Description	Default
`volume_path`	`str \| None`	Full mounted Volume root, e.g. `/Volumes/<catalog>/<schema>/<volume>`.	`None`
`catalog`	`str \| None`	UC catalog name (used when `volume_path` is omitted).	`None`
`schema`	`str \| None`	UC schema name (used when `volume_path` is omitted).	`None`
`volume`	`str \| None`	UC volume name (used when `volume_path` is omitted).	`None`
`prefix`	`str`	Optional sub-path beneath the Volume root (e.g. `"parquet"`).	`''`
`base_path`	`str`	Mount root for Volumes; `/Volumes` inside Databricks.	`'/Volumes'`
`hour_pattern`	`str`	Pattern for the hour-level subpath; tokens `{Y} {m} {d} {H}`. Default `YYYY/MM/DD/HH/`.	`'{Y}/{m}/{d}/{H}/'`
`validate`	`bool`	Warn (`LoaderConfigWarning`) if the resolved root does not exist, so a missing/mistyped mount fails loudly but cheaply.	`True`

load_all_files ¤

load_all_files(
    columns: list[str] | None = None,
    filters: list | None = None,
) -> pd.DataFrame

Load every parquet file under the Volume root (optionally below prefix).

This walks the whole tree; for large Volumes prefer :meth:load_by_time_range or :meth:stream_by_time_range.

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

load_by_time_range ¤

load_by_time_range(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    columns: list[str] | None = None,
    filters: list | None = None,
) -> pd.DataFrame

Load parquet files under the hourly folders within [start, end].

Visits only the YYYY/MM/DD/HH folders in range, so the rest of the Volume is never scanned.

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

stream_by_time_range ¤

stream_by_time_range(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    columns: list[str] | None = None,
    filters: list | None = None,
) -> Iterator[tuple[str, pd.DataFrame]]

Stream parquet frames under the hourly folders within [start, end].

Yields (file_path, DataFrame) one at a time so the full dataset is never held in memory -- the recommended low-memory entry point for Databricks pipelines.

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

load_files_by_time_range_and_uuids ¤

load_files_by_time_range_and_uuids(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    uuid_list: list[str],
    columns: list[str] | None = None,
    filters: list | None = None,
) -> pd.DataFrame

Load parquet files for the given UUIDs within [start, end] hours.

Each hour folder is scanned once and files whose basename matches a requested UUID (<uuid>.parquet, case-insensitive) are read.

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

stream_files_by_time_range_and_uuids ¤

stream_files_by_time_range_and_uuids(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    uuid_list: list[str],
    columns: list[str] | None = None,
    filters: list | None = None,
) -> Iterator[tuple[str, pd.DataFrame]]

Stream parquet frames for the given UUIDs within [start, end] hours.

Yields (file_path, DataFrame) as they are read, keeping memory flat.

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

list_structure ¤

list_structure(
    parquet_only: bool = True, limit: int | None = None
) -> dict[str, list[str]]

List the folders (hours) and files under the configured Volume path.

Parameters:

Name	Type	Description	Default
`parquet_only`	`bool`	If True, only include files ending with `.parquet`.	`True`
`limit`	`int \| None`	Optional cap on number of files collected for quick inspection.	`None`

Returns:

Type	Description
`dict[str, list[str]]`	A dict with `folders` (sorted unique parent folders) and `files`
`dict[str, list[str]]`	(sorted file paths) as strings relative-free full mounted paths.

fetch_data_as_dataframe ¤

fetch_data_as_dataframe(
    start_timestamp: str | Timestamp | None = None,
    end_timestamp: str | Timestamp | None = None,
    columns: list[str] | None = None,
    filters: list | None = None,
) -> pd.DataFrame

Return a combined DataFrame, for Pipeline / DataIntegratorHybrid.

With a start/end pair this delegates to :meth:load_by_time_range (visiting only the in-range hour folders); with no bounds it falls back to :meth:load_all_files.