Data Transformation using Python/PySpark in Lakehouse

Hi datagig 

 The following code is AI generated and unconfirmed but I hope it helps!

# Fabric Notebook: Extract Maturity Excel Sheets -> Summary & Detail Lakehouse tables

import os
import re
import pandas as pd

# ----------------------------
# Config – update these values
# ----------------------------
ROOT_DIR = "/lakehouse/default/Files/X"  # change X to your main folder name under Files
SKIP_FIRST_N_SUBFOLDERS = 7              # skip first 7 subfolders under X
DETAIL_COLUMNS = [
    # TODO: put your 3 column names EXACTLY as they appear in the Detail sheets
    # e.g., "Customer", "Product", "Amount"
]
SUMMARY_TABLE_NAME = "maturity_summary"   # Lakehouse table name for the Summary
DETAIL_TABLE_NAME = "maturity_detail"     # Lakehouse table name for the Detail
OUTPUT_FILES_DIR = "/lakehouse/default/Files/outputs/maturity"  # optional file outputs


# ----------------------------
# Helpers
# ----------------------------
def list_top_level_subfolders(path: str) -> list[str]:
    """Return a sorted list of immediate subfolders under path."""
    if not os.path.exists(path):
        raise FileNotFoundError(f"Root path not found: {path}")
    subfolders = [
        f for f in os.listdir(path)
        if os.path.isdir(os.path.join(path, f))
    ]
    # deterministic ordering: alphabetical
    subfolders.sort()
    return subfolders


def main_folder_tag(folder_name: str) -> str:
    """Tag = part before first '-' if present, else whole folder name."""
    return folder_name.split("-", 1)[0].strip()


def extract_quarter_tag(filename: str) -> str | None:
    """
    Extract 'FYXX Hx' (e.g. FY24 H1) from filename, case-insensitive.
    Accepts optional spaces/underscores/dashes between parts.
    """
    # Normalize
    fname = filename.replace("_", " ").replace("-", " ")
    m = re.search(r"(FY)\s*(\d{2})\s*H\s*([12])", fname, re.IGNORECASE)
    if m:
        yy = m.group(2)
        h = m.group(3)
        return f"FY{yy} H{h}"
    return None


def is_excel_file(path: str) -> bool:
    return path.lower().endswith((".xlsx", ".xls"))


def matches_maturity(filename: str) -> bool:
    return "maturity" in filename.lower()


def select_columns_case_insensitive(df: pd.DataFrame, wanted: list[str]) -> pd.DataFrame:
    """
    Select columns from df matching wanted names case-insensitively.
    Preserves order of wanted.
    """
    if not wanted:
        raise ValueError("DETAIL_COLUMNS is empty. Please set the 3 detail column names.")
    lower_map = {c.lower(): c for c in df.columns}
    selected_actual = []
    missing = []
    for w in wanted:
        key = w.lower()
        if key in lower_map:
            selected_actual.append(lower_map[key])
        else:
            missing.append(w)
    if missing:
        raise KeyError(f"Detail columns not found (case-insensitive): {missing}\nAvailable: {list(df.columns)}")
    return df[selected_actual]


def read_sheet_by_index(file_path: str, sheet_index: int) -> pd.DataFrame | None:
    """
    Read a single sheet by 0-based index.
    Returns None if the sheet index is out of range or if reading fails.
    """
    try:
        if file_path.lower().endswith(".xlsx"):
            df = pd.read_excel(file_path, sheet_name=sheet_index, engine="openpyxl")
        else:
            df = pd.read_excel(file_path, sheet_name=sheet_index, engine="xlrd")
        return df
    except Exception as e:
        # Common reasons: sheet index out of range, bad workbook, etc.
        print(f"  [WARN] Cannot read sheet {sheet_index+1} from {os.path.basename(file_path)}: {e}")
        return None


def ensure_output_dirs():
    try:
        os.makedirs(OUTPUT_FILES_DIR, exist_ok=True)
    except Exception as e:
        print(f"[WARN] Could not create output dir {OUTPUT_FILES_DIR}: {e}")


# ----------------------------
# Processing
# ----------------------------
summary_rows = []  # list of pandas DataFrames
detail_rows = []   # list of pandas DataFrames

subfolders = list_top_level_subfolders(ROOT_DIR)
if len(subfolders) <= SKIP_FIRST_N_SUBFOLDERS:
    print(f"[INFO] Only {len(subfolders)} subfolders under {ROOT_DIR}. Nothing to process after skipping {SKIP_FIRST_N_SUBFOLDERS}.")
else:
    to_process = subfolders[SKIP_FIRST_N_SUBFOLDERS:]
    print(f"[INFO] Found {len(subfolders)} subfolders. Skipping {SKIP_FIRST_N_SUBFOLDERS}, processing {len(to_process)}:")
    for sf in to_process:
        tag = main_folder_tag(sf)
        sf_path = os.path.join(ROOT_DIR, sf)
        print(f"\n[Folder] {sf}  -> tag: '{tag}'")

        # recurse into this subfolder to find matching Excel files
        matched_files = []
        for root, dirs, files in os.walk(sf_path):
            for fn in files:
                if is_excel_file(fn) and matches_maturity(fn):
                    matched_files.append(os.path.join(root, fn))

        if not matched_files:
            print("  [INFO] No matching 'Maturity' Excel files in this folder.")
            continue

        print(f"  [INFO] {len(matched_files)} matching workbook(s) found.")
        for fpath in matched_files:
            fname = os.path.basename(fpath)
            quarter = extract_quarter_tag(fname)
            if not quarter:
                print(f"  [WARN] Could not extract quarter tag from filename '{fname}'. Row will have quarter=None.")
            print(f"  [File] {fname} | quarter: {quarter}")

            # ---- Summary: 3rd sheet (index 2)
            df_sum = read_sheet_by_index(fpath, sheet_index=2)
            if df_sum is not None and not df_sum.empty:
                # Tagging columns
                df_sum["main_folder"] = tag
                df_sum["quarter"] = quarter
                df_sum["source_file"] = fname
                summary_rows.append(df_sum)
            else:
                print("    [INFO] Summary sheet (3rd) missing or empty.")

            # ---- Details: sheets 4..20 (indexes 3..19)
            dfs_detail_this_file = []
            for idx in range(3, 20):
                df_det = read_sheet_by_index(fpath, sheet_index=idx)
                if df_det is None or df_det.empty:
                    continue
                try:
                    # select only desired columns
                    df_sel = select_columns_case_insensitive(df_det, DETAIL_COLUMNS)
                except Exception as e:
                    print(f"    [WARN] Skipping sheet {idx+1} due to column selection error: {e}")
                    continue
                # Tagging columns
                df_sel["main_folder"] = tag
                df_sel["quarter"] = quarter
                df_sel["source_file"] = fname
                dfs_detail_this_file.append(df_sel)

            if dfs_detail_this_file:
                detail_rows.append(pd.concat(dfs_detail_this_file, ignore_index=True))
            else:
                print("    [INFO] No usable Detail sheets (4–20) in this workbook.")

# ----------------------------
# Combine & Save
# ----------------------------
summary_df = pd.concat(summary_rows, ignore_index=True) if summary_rows else pd.DataFrame()
detail_df  = pd.concat(detail_rows,  ignore_index=True) if detail_rows  else pd.DataFrame()

print("\n[RESULT] Combined shapes:")
print(f"  Summary: {summary_df.shape}")
print(f"  Detail : {detail_df.shape}")

# Convert to Spark DataFrames
if summary_df is not None and not summary_df.empty:
    summary_sdf = spark.createDataFrame(summary_df)
    summary_sdf.write.format("delta").mode("overwrite").saveAsTable(SUMMARY_TABLE_NAME)
    print(f"[SAVE] Lakehouse table written: {SUMMARY_TABLE_NAME}")
else:
    print("[SAVE] Summary is empty. No table written.")

if detail_df is not None and not detail_df.empty:
    detail_sdf = spark.createDataFrame(detail_df)
    detail_sdf.write.format("delta").mode("overwrite").saveAsTable(DETAIL_TABLE_NAME)
    print(f"[SAVE] Lakehouse table written: {DETAIL_TABLE_NAME}")
else:
    print("[SAVE] Detail is empty. No table written.")

# Optional: also write files to /Files for ad-hoc access
ensure_output_dirs()
try:
    if summary_df is not None and not summary_df.empty:
        summary_df.to_parquet(os.path.join(OUTPUT_FILES_DIR, "maturity_summary.parquet"), index=False)
    if detail_df is not None and not detail_df.empty:
        detail_df.to_parquet(os.path.join(OUTPUT_FILES_DIR, "maturity_detail.parquet"), index=False)
    print(f"[SAVE] Parquet exports written under {OUTPUT_FILES_DIR}")
except Exception as e:
    print(f"[WARN] Could not write Parquet exports: {e}")

--------------------------------

I hope this helps, please give kudos and mark as solved if it does!

Connect with me on LinkedIn.

Subscribe to my YouTube channel for Fabric/Power Platform related content!