databricks_unity_energy_loader

databricks_unity_energy_loader ¤

DatabricksUnityEnergyLoader ¤

DatabricksUnityEnergyLoader(
    volume_path: str | None = None,
    *,
    catalog: str | None = None,
    schema: str | None = None,
    volume: str | None = None,
    prefix: str = "",
    base_path: str = "/Volumes",
    thousands: str | None = None,
    decimal: str = ".",
    validate: bool = True
)

Load CSV energy timeseries and series metadata governed by Unity Catalog.

Unity Catalog exposes the same CSV energy 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::

<volume_root>/<prefix>/
  .meta/
    series.csv           ← series metadata (tab-separated)
  csv/
    YYYY/MM/DD/
      <series_id>.csv    ← interval readings: time, value columns

All load methods return the standard ts-shape schema::

systime | uuid | value_double | is_delta

where uuid is the series_id (stem of the CSV filename), so the frames 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 CSV directly from that mounted path with pandas.read_csv and 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 day folders in range (csv/YYYY/MM/DD) instead of scanning the whole tree, so a one-day query never lists the whole Volume.
Streaming generator (:meth:stream_by_time_range) yields one series 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.

Off-cluster (outside Databricks) the same code works against any mounted or synced copy of the Volume; if you cannot mount it, use :class:AzureBlobEnergyLoader 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
DatabricksUnityEnergyLoader(
    volume_path="/Volumes/main/plant/energy",
)

# from catalog/schema/volume parts
DatabricksUnityEnergyLoader(
    catalog="main", schema="plant", volume="energy",
)

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 holding the `.meta/` and `csv/` folders.	`''`
`base_path`	`str`	Mount root for Volumes; `/Volumes` inside Databricks.	`'/Volumes'`
`thousands`	`str \| None`	Thousands separator for `pd.read_csv` (default None).	`None`
`decimal`	`str`	Decimal separator for `pd.read_csv` (default ".").	`'.'`
`validate`	`bool`	Warn (`LoaderConfigWarning`) if the resolved root does not exist, so a missing/mistyped mount fails loudly but cheaply.	`True`

load_series_metadata ¤

load_series_metadata() -> pd.DataFrame

Read .meta/series.csv and return as a DataFrame.

id, label_lvl1, label_lvl2, label_lvl3, label_lvl4,

description, unit, hierarchy_lvl1 … hierarchy_lvl6

Returns an empty DataFrame with the expected columns when the file does not exist.

load_by_time_range ¤

load_by_time_range(
    start: Union[str, Timestamp],
    end: Union[str, Timestamp],
    series_ids: list[str] | None = None,
) -> pd.DataFrame

Load all CSV files in csv/YYYY/MM/DD/ for each date in [start, end].

Parameters:

Name	Type	Description	Default
`start`	`Union[str, Timestamp]`	Start date/datetime (inclusive).	required
`end`	`Union[str, Timestamp]`	End date/datetime (inclusive).	required
`series_ids`	`list[str] \| None`	Optional list of series IDs to load. Loads all if None.	`None`

Returns:

Type	Description
`DataFrame`	Standard schema DataFrame: systime \| uuid \| value_double \| is_delta

load_by_series_ids ¤

load_by_series_ids(
    series_ids: list[str],
    start: Union[str, Timestamp] | None = None,
    end: Union[str, Timestamp] | None = None,
) -> pd.DataFrame

Load specific series by ID.

With start/end: visits only the day folders in range and keeps files whose stem matches a requested series_id. Without dates: walks csv/ and filters by stem.

Parameters:

Name	Type	Description	Default
`series_ids`	`list[str]`	Series IDs to load.	required
`start`	`Union[str, Timestamp] \| None`	Optional start date filter.	`None`
`end`	`Union[str, Timestamp] \| None`	Optional end date filter.	`None`

Returns:

Type	Description
`DataFrame`	Standard schema DataFrame: systime \| uuid \| value_double \| is_delta

stream_by_time_range ¤

stream_by_time_range(
    start: Union[str, Timestamp],
    end: Union[str, Timestamp],
    series_ids: list[str] | None = None,
) -> Iterator[tuple[str, pd.DataFrame]]

Stream CSV files one at a time as (series_id, DataFrame) tuples.

Memory-efficient alternative to :meth:load_by_time_range for large date ranges -- the recommended low-memory entry point in pipelines.

Parameters:

Name	Type	Description	Default
`start`	`Union[str, Timestamp]`	Start date/datetime (inclusive).	required
`end`	`Union[str, Timestamp]`	End date/datetime (inclusive).	required
`series_ids`	`list[str] \| None`	Optional series filter.	`None`

Yields:

Type	Description
`tuple[str, DataFrame]`	(series_id, DataFrame) where DataFrame has the standard schema.

list_series ¤

list_series() -> list[str]

List all series IDs present in the Volume by scanning csv/.

Returns:

Type	Description
`list[str]`	Sorted list of unique series ID strings.

fetch_data_as_dataframe ¤

fetch_data_as_dataframe(
    start: Union[str, Timestamp] | None = None,
    end: Union[str, Timestamp] | None = None,
    series_ids: list[str] | 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 day folders); with no bounds it loads every series under csv/.