PySpark Transformations: Code Your Pipeline

PySpark in Fabric

Reading and writing data

Read from a lakehouse

# Read a Delta table
df_orders = spark.read.format("delta").load("Tables/FactOrders")

# Or use the table name directly
df_orders = spark.table("lakehouse.FactOrders")

# Read raw files
df_csv = spark.read.format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("Files/raw/orders/*.csv")

Write to a lakehouse

# Write as Delta table (overwrite)
df_result.write.format("delta") \
    .mode("overwrite") \
    .save("Tables/FactOrdersCleaned")

# Write as Delta table (append)
df_new_rows.write.format("delta") \
    .mode("append") \
    .save("Tables/FactOrders")

# Save as managed table
df_result.write.saveAsTable("lakehouse.FactOrdersCleaned")

Common transformations

Filtering rows

# Keep only completed orders from 2026
df_filtered = df_orders.filter(
    (col("status") == "completed") &   # What's happening: keep only completed
    (year(col("order_date")) == 2026)   # AND only from 2026
)

Selecting and renaming columns

df_clean = df_orders.select(
    col("order_id"),                              # Keep as-is
    col("customer_name").alias("customer"),        # Rename
    col("total_amount").cast("decimal(10,2)"),     # Cast type
    upper(col("country")).alias("country_code")    # Transform + rename
)

Joins (denormalization)

Denormalization in PySpark is a join — combine normalised source tables into wider analytics tables.

# Join orders with customers and products
df_denormalized = df_orders \
    .join(df_customers, df_orders.customer_id == df_customers.customer_id, "left") \
    .join(df_products, df_orders.product_id == df_products.product_id, "left") \
    .select(
        df_orders.order_id,
        df_orders.order_date,
        df_customers.customer_name,      # From customers table
        df_customers.city,               # Flattened into one row
        df_products.product_name,        # From products table
        df_products.category,            # Flattened into one row
        df_orders.quantity,
        df_orders.revenue
    )

What’s happening: Three normalised tables (orders, customers, products) become one wide fact table with customer and product attributes included. This is denormalization.

df_fact = df_batches \
    .join(df_machines, "machine_id", "left") \
    .join(df_products, "product_id", "left") \
    .join(df_factories, "factory_id", "left") \
    .join(df_shifts, "shift_id", "left") \
    .select(
        "batch_id", "production_date",
        "factory_name", "factory_region",
        "machine_type", "product_name", "product_category",
        "shift_name", "units_produced", "units_defective"
    )

Grouping and aggregation

from pyspark.sql.functions import sum, avg, count, max, year, month, year, month

# Revenue by product category per month
df_summary = df_orders \
    .groupBy(
        year("order_date").alias("year"),           # Group by year
        month("order_date").alias("month"),         # and month
        "product_category"                          # and category
    ) \
    .agg(
        sum("revenue").alias("total_revenue"),      # Sum revenue
        count("order_id").alias("order_count"),     # Count orders
        avg("revenue").alias("avg_order_value")     # Average order value
    ) \
    .orderBy("year", "month", "product_category")   # Sort results

Window functions

Window functions calculate values across a set of rows related to the current row — without collapsing them (unlike groupBy).

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, lag, sum as spark_sum

# Running total of revenue per customer, ordered by date
window_spec = Window \
    .partitionBy("customer_id") \
    .orderBy("order_date") \
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)

df_with_running_total = df_orders.withColumn(
    "running_revenue",
    spark_sum("revenue").over(window_spec)   # Running sum within each customer
)

# Previous order date (for calculating days between orders)
df_with_prev = df_orders.withColumn(
    "prev_order_date",
    lag("order_date", 1).over(
        Window.partitionBy("customer_id").orderBy("order_date")
    )
)

Handling nulls

# Drop rows where critical columns are null
df_clean = df_orders.dropna(subset=["customer_id", "order_date"])

# Fill nulls with defaults
df_filled = df_orders.fillna({
    "discount": 0.0,          # Missing discount = no discount
    "shipping_method": "standard"  # Default shipping
})

# Replace specific values
df_fixed = df_orders.withColumn(
    "status",
    when(col("status").isNull(), "unknown")  # Null → "unknown"
    .otherwise(col("status"))                 # Keep existing
)

Writing patterns comparison

Next up: Transform Data with SQL & KQL — use T-SQL and Kusto Query Language for transformations in warehouses and Eventhouses.