) .withColumn("txn_date", to_timestamp(col("InvoiceDate"), "M/d/yyyy H:mm")) .withColumn("unit_price", round(col("UnitPrice"), 2)) # 5. Derive business metrics .withColumn("line_total", round(col("Quantity") * col("UnitPrice"), 2)) .withColumn("product_name", upper(trim(col("Description")))) .withColumn("txn_year", year(col("txn_date"))) .withColumn("txn_month", month(col("txn_date"))) .withColumn("spend_tier", when(col("line_total") >= 500, "Premium") .when(col("line_total") >= 100, "Standard") .otherwise("Entry") ) # 6. Rename to consistent snake_case .withColumnRenamed("InvoiceNo", "txn_id") .withColumnRenamed("StockCode", "sku") .withColumnRenamed("Country", "region") # 7. Drop raw metadata, add processing timestamp .drop("ingest_timestamp", "source_path", "system_origin") .withColumn("processed_at", current_timestamp()) )

Write partitioned by temporal keys for efficient filtering

conformed.write.format("delta").mode("overwrite").partitionBy("txn_year", "txn_month").save(SILVER_PATH)

spark.sql(f""" CREATE TABLE IF NOT EXISTS curated_ecommerce.transactions_clean USING DELTA LOCATION '{SILVER_PATH}' """)

print(f"Silver conformance complete. Rows: {conformed.count():,}")

**Architecture Rationale:**
- Partitioning by `txn_year` and `txn_month` aligns with common BI query patterns, enabling partition pruning.
- Explicit type casting and timestamp parsing prevent silent data corruption downstream.
- Overwriting Silver is acceptable here because the source is Bronze (append-only). If streaming, `MERGE INTO` would be required.

### Phase 4: Gold Aggregation (Business-Ready Metrics)

Gold tables serve pre-computed aggregates optimized for reporting, dashboards, and machine learning features. Logic is isolated from transformation pipelines.

```python
# 03_gold_aggregations.py
from pyspark.sql.functions import sum, count, avg, countDistinct, round

CLEAN_TABLE = "curated_ecommerce.transactions_clean"
GOLD_BASE = "/Volumes/warehouse/analytics_ecommerce"

clean = spark.table(CLEAN_TABLE)

# Gold 1: Regional Revenue Trends
regional_revenue = (
    clean.groupBy("txn_year", "txn_month", "region")
    .agg(
        round(sum("line_total"), 2).alias("gross_revenue"),
        count("txn_id").alias("order_count"),
        round(avg("line_total"), 2).alias("avg_basket_size"),
        countDistinct("customer_id").alias("active_shoppers")
    )
    .orderBy("txn_year", "txn_month", "gross_revenue", ascending=[True, True, False])
)
regional_revenue.write.format("delta").mode("overwrite").save(f"{GOLD_BASE}/regional_revenue")
spark.sql(f"CREATE TABLE IF NOT EXISTS analytics_ecommerce.regional_revenue USING DELTA LOCATION '{GOLD_BASE}/regional_revenue'")

# Gold 2: Product Velocity & Revenue
product_metrics = (
    clean.groupBy("sku", "product_name")
    .agg(
        round(sum("line_total"), 2).alias("total_revenue"),
        sum("Quantity").alias("units_moved"),
        count("txn_id").alias("purchase_frequency"),
        countDistinct("customer_id").alias("buyer_reach"),
        round(avg("unit_price"), 2).alias("realized_price")
    )
    .orderBy("total_revenue", ascending=False)
)
product_metrics.write.format("delta").mode("overwrite").save(f"{GOLD_BASE}/product_metrics")
spark.sql(f"CREATE TABLE IF NOT EXISTS analytics_ecommerce.product_metrics USING DELTA LOCATION '{GOLD_BASE}/product_metrics'")

# Gold 3: Customer Lifetime Value Segments
customer_ltv = (
    clean.groupBy("customer_id", "region")
    .agg(
        round(sum("line_total"), 2).alias("ltv"),
        count("txn_id").alias("total_orders"),
        round(avg("line_total"), 2).alias("avg_order_val"),
        countDistinct("sku").alias("category_breadth")
    )
    .withColumn("segment",
        when(col("ltv") >= 5000, "VIP")
        .when(col("ltv") >= 1000, "Core")
        .when(col("ltv") >= 200,  "Active")
        .otherwise("Dormant")
    )
    .orderBy("ltv", ascending=False)
)
customer_ltv.write.format("delta").mode("overwrite").save(f"{GOLD_BASE}/customer_ltv")
spark.sql(f"CREATE TABLE IF NOT EXISTS analytics_ecommerce.customer_ltv USING DELTA LOCATION '{GOLD_BASE}/customer_ltv'")

print("Gold aggregation pipeline complete.")

Architecture Rationale:

• Gold tables are denormalized for query performance. Joins are pre-computed.
• Separating business logic into Gold allows BI teams to query directly without understanding transformation pipelines.
• Delta's OVERWRITE is safe here because Gold is fully recomputed from Silver on each run.

Pitfall Guide

Production Bundle

Action Checklist

• Define layer boundaries explicitly in pipeline documentation and code comments
• Enable Delta schema evolution (mergeSchema) for Bronze ingestion
• Implement partition pruning strategy aligned with common BI query patterns
• Add ingestion and processing metadata columns for lineage tracking
• Configure spark.sql.adaptive.enabled and spark.sql.adaptive.coalescePartitions for shuffle optimization
• Schedule OPTIMIZE and VACUUM jobs to manage small files and storage retention
• Register all tables in Unity Catalog or workspace schema for access control
• Add data quality checks (null rates, duplicate counts, range validation) between layers

Decision Matrix

Configuration Template

-- Delta Table Optimization & Retention Policy
ALTER TABLE curated_ecommerce.transactions_clean SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true',
  'delta.logRetentionDuration' = 'interval 30 days',
  'delta.deletedRetentionDuration' = 'interval 7 days',
  'delta.appendOnly' = 'false'
);

-- Enable Auto Compaction for Gold Tables
ALTER TABLE analytics_ecommerce.regional_revenue SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true'
);

Quick Start Guide

1. Provision Storage & Catalog: Create three schemas (raw_ecommerce, curated_ecommerce, analytics_ecommerce) and mount a shared volume for Delta tables.
2. Run Bronze Ingestion: Execute the raw CSV ingestion notebook. Verify row counts and metadata columns. Confirm append-only behavior.
3. Execute Silver Conformance: Run the transformation notebook. Validate null removal, deduplication, and partition layout. Check txn_year/txn_month distribution.
4. Generate Gold Aggregates: Execute the aggregation notebook. Query Gold tables via SQL or BI connector. Verify metric consistency against Silver.
5. Schedule & Monitor: Wrap notebooks in a Databricks Workflow. Configure email alerts on row count deltas. Schedule weekly OPTIMIZE and monthly VACUUM jobs.

This architecture transforms raw transactional noise into reliable, query-optimized analytics. By enforcing layer boundaries, leveraging Delta Lake's transactional guarantees, and isolating business logic, teams can scale data products without accumulating technical debt.