databricks_spark_parquet_loader

databricks_spark_parquet_loader ¤

Spark-native Databricks loader for the canonical hourly parquet layout.

Unlike :class:DatabricksUnityParquetLoader -- which reads a FUSE-mounted Unity Catalog Volume directly with pandas.read_parquet -- this loader uses the cluster's Spark session to read the same <base>/YYYY/MM/DD/HH/<uuid>.parquet layout. Use it when you want Spark to do the scan (predicate/column pushdown, distributed read) and only collect to pandas at the end, e.g. inside a Databricks notebook or job.

It generalizes the common notebook idiom::

hours = pd.date_range(start, end, freq="h")
paths = [f"{base}/{h:%Y/%m/%d/%H}/{uuid}.parquet" for h in hours]
sdf = spark.read.option("basePath", base).parquet(*paths)
df = sdf.select("systime", "uuid", "value_integer").toPandas()

into a reusable, parameterized loader: many UUIDs, configurable hour layout, column projection, a Spark filter expression, optional missing-path tolerance, and a choice of Spark or pandas output. The path-building step is a pure, Spark-free method so it can be inspected and unit-tested on its own.

DatabricksSparkParquetLoader ¤

DatabricksSparkParquetLoader(
    base_path: str,
    spark: Any | None = None,
    *,
    hour_pattern: str = "%Y/%m/%d/%H",
    file_template: str = "{uuid}.parquet"
)

Read the canonical hourly parquet layout through a Spark session.

Parameters:

Name	Type	Description	Default
`base_path`	`str`	Root under which the hour folders live, e.g. `/Volumes/main/plant/timeseries` or `abfss://.../timeseries`.	required
`spark`	`Any \| None`	The SparkSession to use. When omitted, the active session is used (`SparkSession.getActiveSession()`); a clear :class:`~ts_shape.errors.LoaderError` is raised at read time if none is available.	`None`
`hour_pattern`	`str`	`strftime` pattern for the hour subfolder. Default `"%Y/%m/%d/%H"` matches the rest of ts-shape.	`'%Y/%m/%d/%H'`
`file_template`	`str`	Per-file name template; `{uuid}` is substituted. Default `"{uuid}.parquet"`.	`'{uuid}.parquet'`

build_paths ¤

build_paths(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    uuids: str | Sequence[str],
    *,
    freq: str = "h"
) -> list[str]

Build the explicit parquet paths covering [start, end] per UUID.

This is the reusable, testable core: it mirrors what Spark will read, with one path per (hour, uuid) pair. Hours are inclusive of both ends.

Parameters:

Name	Type	Description	Default
`start_timestamp`	`str \| Timestamp`	Window start (parsed by `pandas.to_datetime`).	required
`end_timestamp`	`str \| Timestamp`	Window end (inclusive).	required
`uuids`	`str \| Sequence[str]`	A single UUID or a sequence of UUIDs.	required
`freq`	`str`	Folder granularity; defaults to hourly (`"h"`).	`'h'`

Returns:

Type	Description
`list[str]`	A list of fully-qualified parquet paths.

Raises:

Type	Description
`LoaderError`	If no UUIDs are supplied.

load_by_time_range_and_uuids ¤

load_by_time_range_and_uuids(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    uuids: str | Sequence[str],
    *,
    columns: Sequence[str] | None = None,
    filter_expr: str | None = None,
    freq: str = "h",
    ignore_missing: bool = True,
    path_exists: Callable[[str], bool] | None = None,
    as_pandas: bool = True
) -> Any

Read the hourly parquet files for uuids within [start, end].

Parameters:

Name	Type	Description	Default
`start_timestamp`	`str \| Timestamp`	Window start.	required
`end_timestamp`	`str \| Timestamp`	Window end (inclusive).	required
`uuids`	`str \| Sequence[str]`	A single UUID or a sequence of UUIDs.	required
`columns`	`Sequence[str] \| None`	Optional columns to `select` (projection pushdown).	`None`
`filter_expr`	`str \| None`	Optional Spark SQL predicate applied with `.where`, e.g. `"value_integer > 0"`.	`None`
`freq`	`str`	Folder granularity; defaults to hourly.	`'h'`
`ignore_missing`	`bool`	When True, set `spark.sql.files.ignoreMissingFiles` so hours with no file for a UUID are skipped instead of failing.	`True`
`path_exists`	`Callable[[str], bool] \| None`	Optional callable used to pre-filter paths to those that exist (e.g. a wrapper over `dbutils.fs`). Useful when `ignore_missing` is not sufficient for your storage layer.	`None`
`as_pandas`	`bool`	When True (default) return a pandas DataFrame via `toPandas()`; when False return the Spark DataFrame.	`True`

Returns:

Type	Description
`Any`	A pandas DataFrame, or the Spark DataFrame when `as_pandas` is
`Any`	False. An empty pandas DataFrame is returned if no candidate paths
`Any`	remain after `path_exists` filtering.

Raises:

Type	Description
`LoaderError`	If no SparkSession is available.

fetch_data_as_dataframe ¤

fetch_data_as_dataframe(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    uuids: str | Sequence[str],
    *,
    columns: Sequence[str] | None = None,
    filter_expr: str | None = None
) -> pd.DataFrame

Pandas-returning convenience for Pipeline / DataIntegratorHybrid.