lakehouse_parquet_read_as_spark

Public callable

Read a Parquet file from a Fabric lakehouse Files path.

This reads from the lakehouse Files/ area using Spark. If Spark cannot read the original Parquet file because of timestamp precision issues, the helper tries a fallback _tsus path. If that fallback file does not exist, it converts the single local Parquet file from nanosecond to microsecond timestamps and retries the fallback path.

Parameters:

Name	Type	Description	Default
lh	Housepath	Lakehouse path object returned by get_path.	required
relative_path	str	Path to the Parquet file under the lakehouse Files/ folder, without the leading "Files/". For example: "raw/orders/orders_2026.parquet".	required
verbose	bool	Whether to print read and fallback progress.	True
spark_session	object	Spark session to use. If omitted, the helper uses the notebook global spark.	None

Returns:

Type	Description
DataFrame	Spark DataFrame loaded from the original or converted Parquet path.

Raises:

Type	Description
ValueError	If relative_path is not a nested file path.
RuntimeError	If neither the original path nor the converted fallback path can be read successfully.

Examples:

>>> lh_source = get_path("Sandbox", "Source", config=CONFIG)
>>> df = lakehouse_parquet_read_as_spark(
...     lh_source,
...     "raw/orders/orders_2026.parquet",
... )
Notes
-----
Assumes Fabric notebook runtime filesystem conventions for local fallback
conversion paths (``/lakehouse/default/Files/...``).

Source code in src/fabricops_kit/fabric_io.py

def lakehouse_parquet_read_as_spark(lh, relative_path, verbose=True, spark_session=None):
    """Read a Parquet file from a Fabric lakehouse Files path.

    This reads from the lakehouse `Files/` area using Spark. If Spark cannot
    read the original Parquet file because of timestamp precision issues, the
    helper tries a fallback `_tsus` path. If that fallback file does not exist,
    it converts the single local Parquet file from nanosecond to microsecond
    timestamps and retries the fallback path.

    Parameters
    ----------
    lh : Housepath
        Lakehouse path object returned by `get_path`.
    relative_path : str
        Path to the Parquet file under the lakehouse `Files/` folder, without
        the leading `"Files/"`. For example:
        `"raw/orders/orders_2026.parquet"`.
    verbose : bool, default True
        Whether to print read and fallback progress.
    spark_session : object, optional
        Spark session to use. If omitted, the helper uses the notebook global
        `spark`.

    Returns
    -------
    pyspark.sql.DataFrame
        Spark DataFrame loaded from the original or converted Parquet path.

    Raises
    ------
    ValueError
        If `relative_path` is not a nested file path.
    RuntimeError
        If neither the original path nor the converted fallback path can be
        read successfully.

    Examples
    --------
    >>> lh_source = get_path("Sandbox", "Source", config=CONFIG)
    >>> df = lakehouse_parquet_read_as_spark(
    ...     lh_source,
    ...     "raw/orders/orders_2026.parquet",
    ... )
    Notes
    -----
    Assumes Fabric notebook runtime filesystem conventions for local fallback
    conversion paths (``/lakehouse/default/Files/...``).
    """
    if not getattr(lh, "root", None):
        raise ValueError("lh.root is required.")
    if not relative_path:
        raise ValueError("relative_path is required.")

    spark_obj = _get_spark(spark_session)

    relative_path = relative_path.lstrip("/")
    if relative_path.startswith("Files/"):
        relative_path = relative_path[len("Files/") :]

    orig_spark_path = f"Files/{relative_path}"
    lakehouse_prefix = "/lakehouse/default/"
    parts = relative_path.split("/")

    if len(parts) < 2:
        raise ValueError("relative_path should look like folder/file.parquet or folder/subfolder/file.parquet.")

    tsus_dir = parts[:-2] + [parts[-2] + "_tsus"]
    tsus_relative_path = "/".join(tsus_dir + [parts[-1]])
    tsus_spark_path = f"Files/{tsus_relative_path}"

    orig_local_path = f"{lakehouse_prefix}{orig_spark_path}"
    tsus_local_path = f"{lakehouse_prefix}{tsus_spark_path}"

    if verbose:
        print(f"Try Spark read: {orig_spark_path}")

    try:
        df = spark_obj.read.parquet(orig_spark_path)
        _ = df.limit(1).collect()
        if verbose:
            print("SUCCESS: Spark read original path.")
        return df
    except Exception as exc:
        if verbose:
            print(f"Original Parquet read failed. Will try fallback path. Exception: {exc}")

    for try_convert in range(2):
        tag = " after single-file convert" if try_convert else ""

        if verbose:
            print(f"Try Spark read: {tsus_spark_path}{tag}")

        try:
            df = spark_obj.read.parquet(tsus_spark_path)
            _ = df.limit(1).collect()
            if verbose:
                print("SUCCESS: Spark read _tsus path.")
            return df

        except Exception as exc:
            msg = str(exc)
            path_not_found = (
                "[PATH_NOT_FOUND]" in msg
                or "Path does not exist" in msg
                or "No such file or directory" in msg
            )

            if try_convert == 0 and path_not_found:
                if verbose:
                    print("PATH NOT FOUND for _tsus parquet. Will convert one file and retry.")

                try:
                    mssparkutils.fs.mkdirs("/".join(tsus_dir))
                except Exception:
                    pass

                _convert_single_parquet_ns_to_us(
                    local_in_path=orig_local_path,
                    local_out_path=tsus_local_path,
                    verbose=verbose,
                )
            else:
                if verbose:
                    print(f"FAILED: Spark read _tsus path. Exception: {exc}")
                break

    raise RuntimeError("Failed to read from both original and _tsus Parquet paths.")