Implementation Blueprint: Production PySpark Pipeline Framework

# transforms.py - Pure transformation without SparkSession side-effects
import pyspark.sql.functions as F

def calculate_monthly_metrics(df):
    return (
        df.filter(F.col("is_valid") == True)
        .groupBy("customer_id", F.date_trunc("month", "order_date").alias("order_month"))
        .agg(
            F.count("order_id").alias("total_orders"),
            F.round(F.sum("amount"), 2).alias("total_spend")
        )
    )

# test_transforms.py - Fast unit test using Chispa
from chispa.dataframe_comparer import assert_df_equality
from transforms import calculate_monthly_metrics

def test_calculate_monthly_metrics(spark):
    input_data = [
        ("C1", "2026-01-10", 100.50, True),
        ("C1", "2026-01-15", 50.00, True),
        ("C1", "2026-01-20", 30.00, False), # invalid record
    ]
    input_df = spark.createDataFrame(input_data, ["customer_id", "order_date", "amount", "is_valid"])
    result_df = calculate_monthly_metrics(input_df)
    
    assert result_df.count() == 1
    assert result_df.collect()[0]["total_spend"] == 150.50