PySpark Transformations in Microsoft Fabric

The Three-Folder Notebook Architecture

At the heart of this Microsoft Fabric project lies an elegant, scalable notebook architecture. Every transformation follows the same pattern, making it trivial to onboard new tables and achieve production scale. The architecture is organized into three layers:

Layer 1: Constants & Functions Notebook

nb_constants_and_functions is the backbone. It's a shared utility notebook that every other notebook references with %run in its first cell. This ensures all notebooks have access to:

• Column rename mappings: Sourced from a CSV lookup that translates SQL Server ERP column names to standardized warehouse naming
• JDBC helper functions: Execute SQL directly against the SQL Warehouse without needing to manage connections manually
• Watermark logic: Handle incremental loads by tracking the last pipeline runtime and returning sensible defaults (1900-01-01) for first-run scenarios

Layer 2: Common Notebooks (Level 1)

These are the foundational transformation notebooks that read directly from the Landing layer and write to the Curated layer:

• nb_com_drivers_d — Dimensions (driver master data)
• nb_com_trips_header_f — Facts (trip headers)
• nb_com_trips_line_f — Facts (trip line items)

These are independent: each reads from its own source table, validates, transforms, and upserts to the curated layer. They can run in parallel.

Layer 3: Complex Notebooks (Level 2)

nb_cab_trips_f is a Level 2 notebook that depends on the transient output of two Level 1 notebooks. It performs complex joins and business logic.

Why %run First?

Every notebook starts with %run nb_constants_and_functions in cell 1. This:

• Executes the utility notebook in the current Spark session context
• Makes all functions (translate_entity_column, execute_sql_jdbc, get_runtime) available to subsequent cells
• Loads the column rename CSV into memory as the global df variable
• Retrieves a fresh JDBC token from the managed identity

The Constants & Functions Notebook Explained

Let's walk through every cell in nb_constants_and_functions. This is where the magic happens.

Cell 1: Imports & Token Retrieval

from pyspark.sql.functions import col, lit, current_date, to_timestamp, concat, to_date, current_timestamp, upper, lower
import pandas as pd
from datetime import datetime
from notebookutils import mssparkutils

# Retrieve a fresh OAuth token for JDBC connections
token = mssparkutils.credentials.getToken("https://database.windows.net/")

The token is the critical piece. Every JDBC write operation in the warehouse requires authentication. By getting a fresh token at the start of each notebook run, we avoid token expiry issues, even for long-running transformations. The token is valid for ~1 hour, which is typically longer than any single notebook execution.

Cell 2: Column Rename Mapping

# Load column rename mappings from CSV
df = pd.read_csv("/lakehouse/default/Files/Rename Columns/Column_Rename.csv")

# Normalize: strip whitespace and convert to lowercase
df['table_name'] = df['table_name'].str.strip().str.lower()
df['existing_column'] = df['existing_column'].str.strip().str.lower()
df['new_column'] = df['new_column'].str.strip()

def translate_entity_column(vr_ent, vr_column):
    """
    Look up the new column name for a given entity and old column.
    If not found, return the old column name (graceful fallback).
    """
    vr_ent, vr_column = vr_ent.lower(), vr_column.lower()
    try:
        new_column_name = df[(df.table_name == vr_ent) & (df.existing_column == vr_column)].new_column.values[0]
    except:
        new_column_name = vr_column
    return new_column_name

This CSV-based mapping is the secret to scalability. Instead of hardcoding column renames in each notebook, we maintain a single CSV file:

• table_name: The ERP entity (DRIVERSENTITY, TRIPSHEADERENTITY, etc.)
• existing_column: The original column name from SQL Server (ALL_CAPS)
• new_column: The standardized name in the warehouse (lower_snake_case)

When a new column arrives from the source system, we just add one line to the CSV. No code changes. The translate_entity_column() function gracefully falls back to the original name if no mapping exists, so the pipeline won't break if we forget to add a mapping.

Cell 3: JDBC Execution Helper

# JDBC URL for the SQL Warehouse in Production
jdbc_url = "jdbc:sqlserver://<your-workspace>.datawarehouse.fabric.microsoft.com:1433;database=dwh_taxi_business_prod;encrypt=true;trustServerCertificate=false"

def execute_sql_jdbc(sql_text, jdbc_url, token):
    """
    Execute SQL directly in the SQL Warehouse using JDBC.
    No Spark DataFrame involved — pure JDBC connection.
    """
    jvm = spark._sc._jvm
    props = jvm.java.util.Properties()
    props.setProperty("accessToken", token)

    conn = jvm.java.sql.DriverManager.getConnection(jdbc_url, props)
    stmt = conn.createStatement()
    stmt.execute(sql_text)
    stmt.close()
    conn.close()

This function is critical. It uses the Spark JVM to execute arbitrary SQL in the SQL Warehouse. Why not use spark.sql()? Because spark.sql() only works with Lakehouse tables. The SQL Warehouse is a separate compute engine. To write to it from Spark, we need JDBC.

The token-based authentication (OAuth 2.0) is more secure than username/password and avoids storing credentials in notebooks. The managed identity of the workspace automatically has permissions on the warehouse.

Cell 4: Watermark Helper Function

def get_runtime(value, load_type):
    """
    Return the effective runtime for incremental load.
    - If load_type is 'FULL', always return 1900-01-01 (full reload)
    - If value is None or empty, return 1900-01-01 (first run ever)
    - Otherwise, return the stored value (incremental load from that datetime)
    """
    if load_type and load_type.lower() == "full":
        return datetime(1900, 1, 1, 0, 0, 0)
    if value is None or value == "":
        return datetime(1900, 1, 1, 0, 0, 0)
    return value

This is elegant. Instead of handling None checks all over the codebase, we centralize it. Returning 1900-01-01 is a trick: any timestamp in your data will be after 1900, so the filter WHERE last_modified_at >= 1900-01-01 effectively becomes "load everything."

This design means:

• First run ever: Value is NULL → return 1900-01-01 → full load
• Incremental run: Value is "2026-04-15 14:30:00" → return that → load only new data
• Manual full reload: Admin sets load_type to "FULL" → return 1900-01-01 → full load

The Common Notebook: nb_com_drivers_d (Step by Step)

Now let's walk through a concrete Level 1 notebook. nb_com_drivers_d reads driver master data from the landing layer, validates it, transforms it, and upserts it to the curated layer. Every Level 1 notebook follows this same 12-step pattern.

Step 1: Read from Landing

%run nb_constants_and_functions

com_drivers_d_landing = spark.read.table("lh_taxi_business.cab_landing.DRIVERSENTITY")

The first non-comment cell is always the %run. The second cell reads the landing table. In this project, landing tables are stored in the Lakehouse's delta format, so we use spark.read.table().

Step 2: Count Check & Early Exit

try:
    com_drivers_d_landing_count = com_drivers_d_landing.limit(1).count()
    if not (com_drivers_d_landing_count > 0):
        notebookutils.notebook.exit({'error': "Staging Count: " + str(com_drivers_d_landing_count)})
except Exception as e:
    print("Exception: ", e)
    notebookutils.notebook.exit({'error': "Exception: " + str(e)})

This is a critical optimization. limit(1).count() checks if at least one row exists without scanning the entire table. If the table is empty (no new data since last run), we exit cleanly, saving Spark compute.

The try/except ensures that if something goes wrong during the existence check, the notebook exits with an error message instead of proceeding with an exception. This is better than a silent failure.

Step 3: Column Rename

colList = com_drivers_d_landing.columns
entity = "DRIVERSENTITY"

for c in colList:
    com_drivers_d_landing = com_drivers_d_landing.withColumnRenamed(c, translate_entity_column(entity, c))

For each column in the landing table, we look up its new name using the translate_entity_column() function. If a mapping exists, we rename it. If not, it keeps the original name (graceful fallback). This is how we convert from SQL Server's ALL_CAPS to warehouse's lower_snake_case.

Step 4: Add Warehousing Metadata Columns

com_drivers_d_landing = com_drivers_d_landing \
    .withColumn("integration_id", concat(col("driver_id"))) \
    .withColumn("data_source", lit("SQLServer")) \
    .withColumn("w_insert_date", current_timestamp()) \
    .withColumn("w_modified_date", current_timestamp()) \
    .withColumn("etl_created_by", lit("rohit.ram")) \
    .withColumn("etl_updated_by", lit("rohit.ram"))

These metadata columns are crucial for data stewardship and audit trails:

• integration_id: The natural key used for upsert matching. For drivers, it's the driver_id. For fact tables, it might be composite (e.g., trip_id~trip_line_id)
• data_source: Track where this record came from. "SQLServer" for this landing, could be "API" or "CRM" for another
• w_insert_date / w_modified_date: Timestamps for audit. The "w_" prefix means "warehouse" (not from source)
• etl_created_by / etl_updated_by: Data stewardship: who last touched this record

Step 5: Reorder Columns (Aesthetics & Convention)

original_cols = com_drivers_d_landing.columns
first_cols = ["integration_id", "data_source"]
last_cols = ["w_insert_date", "w_modified_date", "etl_created_by", "etl_updated_by"]

[original_cols.remove(i) for i in (first_cols + last_cols)]

new_cols = first_cols + original_cols + last_cols
com_drivers_d_landing = com_drivers_d_landing.select(new_cols)

Column order matters for readability and debugging. We always put the key columns first (integration_id, data_source), then business columns, then audit columns at the end. This is a production convention.

Step 6: Write to Transient Layer (Temporary Storage)

com_drivers_d_landing.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("cab_transient.com_drivers_d")

The transient layer is a Lakehouse schema that holds intermediate data between landing and curated. It's temporary (can be dropped/recreated daily). We write it in overwrite mode because each run replaces the previous run's transient data.

Step 7: Read from Transient (Preparation for Curated Upsert)

com_drivers_d_transient = spark.read.table("lh_taxi_business.cab_transient.com_drivers_d")

We read back from transient. This seems redundant (we just wrote it), but it's a safety check: if the write failed silently, we'll catch it when the read fails.

Step 8: THE KEY STEP — Upsert to Curated Layer with DeltaTable.merge()

from delta.tables import DeltaTable

if spark.catalog.tableExists("cab_curated.com_drivers_d"):
    delta_table = DeltaTable.forName(spark, "cab_curated.com_drivers_d")
    print("Started Upsert")

    delta_table.alias("t").merge(
        com_drivers_d_transient.alias("s"),
        "t.integration_id = s.integration_id"
    ).whenMatchedUpdateAll() \
     .whenNotMatchedInsertAll() \
     .execute()

    print("Completed Upsert")
else:
    com_drivers_d_transient.write.format("delta").mode("overwrite").saveAsTable("cab_curated.com_drivers_d")
    print("Table is Created")

This is the heart of the medallion architecture. The DeltaTable.merge() operation is an SQL MERGE statement implemented in Spark:

• First run (table doesn't exist): Create the table with the transient data using overwrite mode
• Subsequent runs (table exists): MERGE using integration_id as the match key:
  • If a driver already exists (integration_id match) → UPDATE all columns
  • If a driver is new → INSERT all columns

This ensures idempotency. If you run this notebook twice on the same data, you get the same result. No duplicate keys, no stale data.

Step 9: Write to SQL Warehouse Staging Table (via JDBC)

com_drivers_d_curated = spark.read.table("lh_taxi_business.cab_curated.com_drivers_d")

com_drivers_d_curated.write \
    .format("jdbc") \
    .option("url", jdbc_url) \
    .option("dbtable", "[staging].[com_drivers_d_stg]") \
    .option("accessToken", token) \
    .mode("append") \
    .save()

We read the curated data and write it to the SQL Warehouse's staging table using JDBC. Note: mode("append"). We don't overwrite because there might be data from parallel runs. The staging table is temporary and will be merged into the main table in the next step.

Step 10: MERGE into EDW Table (Dynamically Generated SQL)

columns = com_drivers_d_curated.columns
key_column = "integration_id"
update_columns = [c for c in columns if c != key_column]

update_set = ",\n    ".join([f"target.{c} = source.{c}" for c in update_columns])
insert_cols = ", ".join(columns)
insert_vals = ", ".join([f"source.{c}" for c in columns])

merge_sql = f"""
MERGE [edw].[com_drivers_d] AS target
USING [staging].[com_drivers_d_stg] AS source
ON target.{key_column} = source.{key_column}
WHEN MATCHED THEN UPDATE SET
    {update_set}
WHEN NOT MATCHED THEN
    INSERT ({insert_cols})
    VALUES ({insert_vals});
"""

execute_sql_jdbc(merge_sql, jdbc_url, token)

This is brilliant: the MERGE SQL is dynamically generated from the DataFrame schema. If tomorrow you add a new column to the source system and the CSV mapping is updated, this merge SQL automatically includes it. No hardcoded column lists. Zero maintenance.

Step 11: Cleanup Staging Table

truncate_sql = "DELETE FROM [staging].[com_drivers_d_stg]"
execute_sql_jdbc(truncate_sql, jdbc_url, token)

After a successful merge, we clean up the staging table so it's ready for the next run. This prevents stale data from being merged again.

The Complex Notebook: nb_cab_trips_f (Multi-Source Join)

This Level 2 notebook demonstrates how to handle dependent tables and complex joins. It reads from two transient tables (trips_header and trips_line) and joins them with watermark filters.

Reading the Config Table for Watermarks

# Read the config table that tracks last pipeline runtimes
landing_control_tbl = spark.read.table("dwh_taxi_business_dev.config.landing_control_tbl")

# Convert to a dictionary for fast lookup
landing_control_tbl = {
    row["destination_table_name"]: {
        "runtime": row["last_pipeline_runtime"],
        "load_type": row["load_type"]
    }
    for row in landing_control_tbl.collect()
}

Applying Watermark Filters from Dependent Tables

# Get the last runtime for the TRIPS HEADER source
trips_header_runtime = landing_control_tbl.get("TRIPSHEADERENTITY")
com_trips_header_f_last_pipeline_runtime = get_runtime(
    trips_header_runtime.get("runtime"),
    trips_header_runtime.get("load_type")
)

# Read transient data and filter to only new records since last run
com_trips_header_f_transient = spark.read.table("lh_taxi_business.cab_transient.com_trips_header_f") \
    .filter(col("last_modified_at") >= lit(str(com_trips_header_f_last_pipeline_runtime)))

# Do the same for trips line items
trips_line_runtime = landing_control_tbl.get("TRIPSLINEENTITY")
com_trips_line_f_last_pipeline_runtime = get_runtime(
    trips_line_runtime.get("runtime"),
    trips_line_runtime.get("load_type")
)

com_trips_line_f_transient = spark.read.table("lh_taxi_business.cab_transient.com_trips_line_f") \
    .filter(col("last_modified_at") >= lit(str(com_trips_line_f_last_pipeline_runtime)))

Preparing for Join (Column Renaming to Avoid Conflicts)

# Rename key columns to avoid ambiguity in the join
com_trips_header_f_transient = com_trips_header_f_transient \
    .withColumnRenamed("integration_id", "trips_header_integration_id") \
    .withColumnRenamed("trip_id", "trips_header_trip_id")

# Drop common columns from line items (keep only line-specific columns)
inner_left_join_cols = list(set(com_trips_line_f_transient.columns) - set(com_trips_header_f_transient.columns))
com_trips_line_f_transient = com_trips_line_f_transient.select(inner_left_join_cols)

The Join

# Left join: keep all header rows, even if they have no line items
cab_trips_f = com_trips_header_f_transient.join(
    com_trips_line_f_transient,
    com_trips_header_f_transient.trips_header_trip_id == com_trips_line_f_transient.trip_id,
    "left"
)

# Create a composite integration_id (header_id ~ line_id)
cab_trips_f = cab_trips_f.withColumn("integration_id",
    concat(col("trips_header_integration_id"), lit("~"), col("trips_line_integration_id"))
)

The composite integration_id (using "~" as a separator) is a production pattern. It uniquely identifies each fact row across multiple source tables. The left join ensures we keep trips even if they have no line items (canceled trips, for example).

Why This Architecture Is Production-Ready

This architecture has been battle-tested at scale. Here's why it works:

1. Standardized Pattern = Easy Scaling

Every Level 1 notebook follows the exact same 11-step pattern. To add a new table to the pipeline:

1. Copy an existing Level 1 notebook
2. Change the landing table name
3. Change the entity name for the column rename
4. Change the key column for the upsert
5. Done. 20 minutes to production.

2. Idempotency Guaranteed

The integration_id-based merge ensures that running the pipeline 10 times on the same data produces the same result. No duplicate keys, no data loss. This is critical for reliability.

3. Audit Trail Built-In

Every row has data_source, w_insert_date, w_modified_date, and etl_created_by. Data governance teams can trace any value back to its source and timestamp.

4. Staging Decouples Concerns

By writing to staging before merging, we avoid table locks. Large datasets can be written without blocking reads.

Key Takeaways

• %run first: Every notebook starts with %run nb_constants_and_functions to share utilities and get a fresh token
• Column mapping via CSV: Maintenance-free column renaming across the pipeline
• Watermark pattern: CDC watermarks in a config table enable incremental loads and dependent notebook logic
• DeltaTable.merge(): The heart of idempotent upserts. First run creates, subsequent runs merge
• Transient layer: Decouples transformation from curated writes, enabling safe retry logic
• Dynamic MERGE SQL: Column-driven DDL means new columns automatically flow through the pipeline
• Staging in SQL Warehouse: JDBC writes to staging, then SQL MERGE to main tables, preventing locks

Rohit Ram

Senior Data Analyst · LinkedIn

Building production-grade data platforms on Microsoft Fabric. This series documents a real project — every SQL query, every design decision, every challenge encountered and solved.