Module 8: Data Transformation & Modeling

%sql
CREATE OR REPLACE TABLE bronze_orders AS
SELECT *, current_timestamp() AS _ingested_at, 'orders_raw' AS _source
FROM orders_raw

bronze = spark.table("bronze_orders")

w = Window.partitionBy("order_id").orderBy(F.desc("_ingested_at"))
silver = (
    bronze
    .withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")
    .withColumn("status", F.lower(F.trim("status")))
    .withColumn("order_date", F.to_date("order_ts"))
    .withColumn("is_valid", F.col("customer_id").isNotNull()
                            & F.col("quantity").isNotNull() & (F.col("quantity") > 0))
    .drop("_source")
)
silver.write.mode("overwrite").saveAsTable("silver_orders")
print(spark.table("silver_orders").count())      # 2000

(spark.table("silver_orders").filter("NOT is_valid")
      .write.mode("overwrite").saveAsTable("silver_quarantine"))
print(spark.table("silver_quarantine").count())  # 35: 20 no customer + 15 no quantity

gold = (
    spark.table("silver_orders").filter("is_valid AND status IN ('completed', 'shipped')")
    .join(spark.table("products"), "product_id")
    .groupBy("order_date")
    .agg(F.count("*").alias("orders"),
         F.round(F.sum(F.col("quantity") * F.col("unit_price")), 2).alias("revenue"))
)
gold.write.mode("overwrite").saveAsTable("gold_daily_revenue")

bronze = spark.table("bronze_orders")

cleaned = (
    bronze
    .withColumn("status", F.lower(F.trim("status")))                   # consistent text
    .withColumn("order_date", F.to_date("order_ts"))                   # right type
    .withColumn("quantity", F.col("quantity").cast("int"))
    .withColumn("customer_id", F.coalesce("customer_id", F.lit(-1)))   # explicit unknown
)
display(cleaned.limit(5))

b = spark.table("bronze_orders")
print(b.count(), b.select("order_id").distinct().count())     # 2040 2000

# exact duplicates
exact = b.dropDuplicates(["order_id"])

# latest by ingestion time, deterministic
w = Window.partitionBy("order_id").orderBy(F.desc("_ingested_at"), F.desc("order_ts"))
latest = b.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")
print(exact.count(), latest.count())                          # 2000 2000

rules = {
    "customer_id is present": F.col("customer_id").isNotNull(),
    "quantity is present":    F.col("quantity").isNotNull(),
    "quantity is positive":   F.col("quantity") > 0,
    "status is known":        F.col("status").isin("completed", "shipped", "cancelled", "returned"),
    "order_ts is present":    F.col("order_ts").isNotNull(),
}

df = spark.table("bronze_orders")
report = df.select(
    [F.sum(F.when(~cond | cond.isNull(), 1).otherwise(0)).alias(name) for name, cond in rules.items()]
)
report.show(truncate=False)

%sql
ALTER TABLE silver_orders ADD CONSTRAINT status_known
CHECK (status IN ('completed', 'shipped', 'cancelled', 'returned'))

# new rows arrive in bronze
spark.sql("""
  INSERT INTO bronze_orders
  SELECT order_id + 5000 AS order_id, customer_id, product_id, quantity, status, order_ts,
         current_timestamp() AS _ingested_at, 'orders_raw' AS _source
  FROM orders_raw WHERE order_id <= 10
""")

# silver picks up only rows newer than what it already has
last = spark.table("silver_orders").agg(F.max("_ingested_at")).first()[0]
new = spark.table("bronze_orders").filter(F.col("_ingested_at") > last)
print(new.count())                                                    # 10

new_clean = (
    new.withColumn("status", F.lower(F.trim("status")))
       .withColumn("order_date", F.to_date("order_ts"))
       .withColumn("is_valid", F.col("customer_id").isNotNull()
                               & F.col("quantity").isNotNull() & (F.col("quantity") > 0))
       .drop("_source")
)
new_clean.createOrReplaceTempView("new_silver")
spark.sql("""
  MERGE INTO silver_orders t USING new_silver s ON t.order_id = s.order_id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *
""")
print(spark.table("silver_orders").count())                           # 2010

# streaming alternative: reads only new bronze rows each run
(spark.readStream.table("bronze_orders")
      .writeStream.trigger(availableNow=True)
      .option("checkpointLocation", f"/Volumes/{catalog}/dbx_course/raw/_checkpoints/silver_stream")
      .toTable("silver_orders_stream"))

dim = spark.table("customers").select(
    F.xxhash64("customer_id").alias("customer_sk"), "customer_id", "name", "country", "signup_date"
)
display(dim.limit(3))

%sql
CREATE OR REPLACE TABLE dim_product_id (
  product_sk   BIGINT GENERATED ALWAYS AS IDENTITY,
  product_id   INT,
  product_name STRING,
  category     STRING
)

%sql
INSERT INTO dim_product_id (product_id, product_name, category)
SELECT product_id, product_name, category FROM products

# dimensions
spark.table("customers").select(
    F.xxhash64("customer_id").alias("customer_sk"), "customer_id", "name", "country", "signup_date"
).write.mode("overwrite").saveAsTable("dim_customer")

spark.table("products").select(
    F.xxhash64("product_id").alias("product_sk"), "product_id", "product_name", "category", "unit_price"
).write.mode("overwrite").saveAsTable("dim_product")

spark.sql("""
  SELECT CAST(date_format(d, 'yyyyMMdd') AS INT) AS date_key, d AS date,
         year(d) AS year, month(d) AS month, date_format(d, 'EEEE') AS weekday,
         dayofweek(d) IN (1, 7) AS is_weekend
  FROM (SELECT explode(sequence(DATE'2025-01-01', DATE'2025-12-31', INTERVAL 1 DAY)) AS d)
""").write.mode("overwrite").saveAsTable("dim_date")
print(spark.table("dim_date").count())          # 365

# fact: one row per valid order
(
    spark.table("silver_orders").filter("is_valid")
    .join(spark.table("products").select("product_id", "unit_price"), "product_id")
    .select(
        "order_id",
        F.xxhash64("customer_id").alias("customer_sk"),
        F.xxhash64("product_id").alias("product_sk"),
        F.date_format("order_date", "yyyyMMdd").cast("int").alias("date_key"),
        "quantity", "unit_price", (F.col("quantity") * F.col("unit_price")).alias("revenue"), "status",
    )
    .write.mode("overwrite").saveAsTable("fact_orders")
)

(spark.table("customers")
      .select("customer_id", "name", "country",
              F.lit("2024-01-01").cast("date").alias("valid_from"),
              F.lit(None).cast("date").alias("valid_to"),
              F.lit(True).alias("is_current"))
      .write.mode("overwrite").saveAsTable("dim_customer_scd2"))

changes = spark.createDataFrame(
    [(1, "Customer 1", "XX"), (2, "Customer 2", "XX"), (201, "Customer 201", "NL")],
    "customer_id INT, name STRING, country STRING",
).withColumn("change_date", F.lit("2026-10-19").cast("date"))
changes.createOrReplaceTempView("changes")

%sql
MERGE INTO dim_customer_scd2 AS t
USING (
  -- rows that match on the key: close the old version
  SELECT c.customer_id AS merge_key, c.* FROM changes c
  UNION ALL
  -- changed rows again with a NULL key: they never match, so they are inserted
  SELECT NULL AS merge_key, c.*
  FROM changes c JOIN dim_customer_scd2 d
    ON c.customer_id = d.customer_id AND d.is_current AND c.country <> d.country
) AS s
ON t.customer_id = s.merge_key AND t.is_current = true
WHEN MATCHED AND s.country <> t.country THEN
  UPDATE SET is_current = false, valid_to = s.change_date
WHEN NOT MATCHED THEN
  INSERT (customer_id, name, country, valid_from, valid_to, is_current)
  VALUES (s.customer_id, s.name, s.country, s.change_date, NULL, true)

t = spark.table("dim_customer_scd2")
print(t.count(), t.filter("is_current").count())     # 205 201
display(t.filter("customer_id IN (1, 201)").orderBy("customer_id", "valid_from"))

fact = spark.table("fact_orders")
dim_p = spark.table("dim_product")
dim_c = spark.table("dim_customer")

by_cat_country = (
    fact.filter("status IN ('completed', 'shipped')")
        .join(dim_p, "product_sk").join(dim_c, "customer_sk")
        .groupBy("category", "country")
        .agg(F.round(F.sum("revenue"), 2).alias("revenue"), F.count("*").alias("orders"))
)
display(by_cat_country.orderBy(F.desc("revenue")))

%sql
SELECT category, country, ROUND(SUM(revenue), 2) AS revenue
FROM fact_orders f
JOIN dim_product p USING (product_sk)
JOIN dim_customer c USING (customer_sk)
GROUP BY ROLLUP (category, country)
ORDER BY category, country

# a late order: belongs to 2025-02-03 but arrives now
late = spark.createDataFrame(
    [(9001, 5, 7, 2, "completed", "2025-02-03 10:00:00")],
    "order_id INT, customer_id INT, product_id INT, quantity INT, status STRING, order_ts STRING",
).withColumn("order_ts", F.to_timestamp("order_ts")) \
 .withColumn("_ingested_at", F.current_timestamp()) \
 .withColumn("_source", F.lit("late"))
late.write.mode("append").saveAsTable("bronze_orders")

# recompute only the affected days in gold
affected = [r[0] for r in late.select(F.to_date("order_ts")).distinct().collect()]
print(affected)

def build_gold_daily(start, end):
    """Rebuild gold_daily_revenue for [start, end] and nothing else."""
    df = (
        spark.table("fact_orders")
        .filter("status IN ('completed', 'shipped')")
        .join(spark.table("dim_date"), "date_key")
        .filter(F.col("date").between(start, end))
        .groupBy(F.col("date").alias("order_date"))
        .agg(F.count("*").alias("orders"), F.round(F.sum("revenue"), 2).alias("revenue"))
    )
    (df.write.mode("overwrite")
       .option("replaceWhere", f"order_date BETWEEN '{start}' AND '{end}'")
       .saveAsTable("gold_daily_revenue"))


build_gold_daily("2025-01-01", "2025-03-31")      # a quarter
build_gold_daily("2025-01-01", "2025-03-31")      # same again: same result

bronze_rows   = spark.table("bronze_orders").count()
bronze_keys   = spark.table("bronze_orders").select("order_id").distinct().count()
silver_rows   = spark.table("silver_orders").count()
quarantine    = spark.table("silver_quarantine").count()
fact_rows     = spark.table("fact_orders").count()

print("bronze distinct keys == silver rows:", bronze_keys == silver_rows)
print("silver == fact + quarantine:        ", silver_rows == fact_rows + quarantine)

%sql
-- keys in silver that are not in the fact (they must be the quarantined ones)
SELECT order_id FROM silver_orders
EXCEPT
SELECT order_id FROM fact_orders