Skip to content

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. The hour folders are UTC: a tz-aware bound is converted to UTC first and a naive bound is read as UTC.
  • 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,
    *,
    errors: ErrorsMode = "warn"
) -> 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
errors ErrorsMode

"warn" (default) skips a file that cannot be read or parsed and emits one :class:~ts_shape.errors.DataQualityWarning naming every skipped file; "raise" raises :class:~ts_shape.errors.LoaderError instead.

 'warn'

load_by_time_range ¤

load_by_time_range(
    start: str | Timestamp,
    end: str | Timestamp,
    columns: list[str] | None = None,
    filters: list | None = None,
    *,
    trim_to_range: bool = True,
    errors: ErrorsMode = "warn"
) -> 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
start str | Timestamp

Start of the range (anything pd.to_datetime accepts). Every hour folder that overlaps [start, end] is read in full (the start is floored to the hour); see trim_to_range.

 required
end str | Timestamp

End of the range, inclusive (anything pd.to_datetime accepts).

 required
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
trim_to_range bool

If True (default), drop rows whose systime falls outside [start, end] (both bounds inclusive). Frames without a systime column are left as read. False keeps whole hour folders.

 True
errors ErrorsMode

"warn" (default) skips a file that cannot be read or parsed and emits one :class:~ts_shape.errors.DataQualityWarning naming every skipped file; "raise" raises :class:~ts_shape.errors.LoaderError instead.

 'warn'

stream_by_time_range ¤

stream_by_time_range(
    start: str | Timestamp,
    end: str | Timestamp,
    columns: list[str] | None = None,
    filters: list | None = None,
    *,
    trim_to_range: bool = True,
    errors: ErrorsMode = "warn"
) -> 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
start str | Timestamp

Start of the range (anything pd.to_datetime accepts). Every hour folder that overlaps [start, end] is read in full (the start is floored to the hour); see trim_to_range.

 required
end str | Timestamp

End of the range, inclusive (anything pd.to_datetime accepts).

 required
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
trim_to_range bool

If True (default), drop rows whose systime falls outside [start, end] (both bounds inclusive). Frames without a systime column are left as read. False keeps whole hour folders.

 True
errors ErrorsMode

"warn" (default) skips a file that cannot be read or parsed and emits one :class:~ts_shape.errors.DataQualityWarning naming every skipped file; "raise" raises :class:~ts_shape.errors.LoaderError instead.

 'warn'

load_files_by_time_range_and_uuids ¤

load_files_by_time_range_and_uuids(
    start: str | Timestamp,
    end: str | Timestamp,
    uuid_list: list[str],
    columns: list[str] | None = None,
    filters: list | None = None,
    *,
    trim_to_range: bool = True,
    errors: ErrorsMode = "warn"
) -> 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
start str | Timestamp

Start of the range (anything pd.to_datetime accepts). Every hour folder that overlaps [start, end] is read in full (the start is floored to the hour); see trim_to_range.

 required
end str | Timestamp

End of the range, inclusive (anything pd.to_datetime accepts).

 required
uuid_list list[str]

Signal UUIDs to load; each matches a <uuid>.parquet file name, case-insensitively.

 required
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
trim_to_range bool

If True (default), drop rows whose systime falls outside [start, end] (both bounds inclusive). Frames without a systime column are left as read. False keeps whole hour folders.

 True
errors ErrorsMode

"warn" (default) skips a file that cannot be read or parsed and emits one :class:~ts_shape.errors.DataQualityWarning naming every skipped file; "raise" raises :class:~ts_shape.errors.LoaderError instead.

 'warn'

stream_files_by_time_range_and_uuids ¤

stream_files_by_time_range_and_uuids(
    start: str | Timestamp,
    end: str | Timestamp,
    uuid_list: list[str],
    columns: list[str] | None = None,
    filters: list | None = None,
    *,
    trim_to_range: bool = True,
    errors: ErrorsMode = "warn"
) -> 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
start str | Timestamp

Start of the range (anything pd.to_datetime accepts). Every hour folder that overlaps [start, end] is read in full (the start is floored to the hour); see trim_to_range.

 required
end str | Timestamp

End of the range, inclusive (anything pd.to_datetime accepts).

 required
uuid_list list[str]

Signal UUIDs to load; each matches a <uuid>.parquet file name, case-insensitively.

 required
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
trim_to_range bool

If True (default), drop rows whose systime falls outside [start, end] (both bounds inclusive). Frames without a systime column are left as read. False keeps whole hour folders.

 True
errors ErrorsMode

"warn" (default) skips a file that cannot be read or parsed and emits one :class:~ts_shape.errors.DataQualityWarning naming every skipped file; "raise" raises :class:~ts_shape.errors.LoaderError instead.

 'warn'

list_structure ¤

list_structure(
    only_parquet: 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
only_parquet 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: str | Timestamp | None = None,
    end: str | Timestamp | None = None,
    columns: list[str] | None = None,
    filters: list | None = None,
    *,
    trim_to_range: bool = True,
    errors: ErrorsMode = "warn"
) -> 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 only one bound, the range is open on the other side: every hour folder at or after the start hour (or at or before the end hour) is read. With no bounds it falls back to :meth:load_all_files.

Parameters:

Name Type Description Default
start str | Timestamp | None

Start of the range; None leaves the start open. Bounds select whole hour folders (the start is floored to the hour); see trim_to_range.

 None
end str | Timestamp | None

End of the range, inclusive; None leaves the end open.

 None
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
trim_to_range bool

If True (default), drop rows whose systime falls outside the given bounds (both inclusive). False keeps whole hour folders.

 True
errors ErrorsMode

"warn" (default) skips a file that cannot be read or parsed and emits one :class:~ts_shape.errors.DataQualityWarning naming every skipped file; "raise" raises :class:~ts_shape.errors.LoaderError instead.

 'warn'