Ingest and transform data

from delta.tables import DeltaTable
from pyspark.sql.functions import col, lit, current_date

dim = DeltaTable.forName(spark, "gold_dim_customer")

# Step 1 — close out rows whose tracked attributes changed
(dim.alias("t")
   .merge(updates.alias("s"),
          "t.customer_id = s.customer_id AND t.is_current = true")
   .whenMatchedUpdate(
       condition="t.address <> s.address OR t.segment <> s.segment",
       set={"is_current": lit(False), "valid_to": current_date()})
   .execute())

# Step 2 — insert the new current versions
(dim.alias("t")
   .merge(updates.alias("s"),
          "t.customer_id = s.customer_id AND t.is_current = true")
   .whenNotMatchedInsert(values={
       "customer_id": "s.customer_id",
       "address":     "s.address",
       "segment":     "s.segment",
       "valid_from":  "current_date()",
       "valid_to":    "null",
       "is_current":  "true"})
   .execute())

from pyspark.sql.functions import col, year, month, sum as _sum, count

df = spark.read.format("delta").load("Tables/bronze_sales")

clean = (df
    .filter(col("amount") > 0)
    .withColumn("order_year", year("order_date"))
    .withColumn("order_month", month("order_date"))
    .dropDuplicates(["order_id"]))

clean.write.format("delta").mode("overwrite") \
     .option("overwriteSchema", "true") \
     .saveAsTable("silver_sales")

# Group and aggregate
agg = (clean.groupBy("order_year", "order_month", "region")
            .agg(_sum("amount").alias("total_amount"),
                 count("*").alias("order_count")))

-- CTAS is the preferred bulk-transform pattern in a Fabric Warehouse
CREATE TABLE gold.monthly_sales AS
SELECT
    YEAR(order_date)  AS sale_year,
    MONTH(order_date) AS sale_month,
    region,
    SUM(amount)       AS total_amount,
    COUNT_BIG(*)      AS order_count
FROM   silver.sales
WHERE  amount > 0
GROUP BY YEAR(order_date), MONTH(order_date), region;

SensorReadings
| where Timestamp > ago(7d)
| summarize AvgTemp = avg(Temperature),
            MaxTemp = max(Temperature),
            Readings = count()
    by bin(Timestamp, 1h), DeviceId
| order by Timestamp desc

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, col, coalesce, lit

w = Window.partitionBy("order_id").orderBy(col("modified_at").desc())

latest = (df.withColumn("rn", row_number().over(w))
            .filter(col("rn") == 1)
            .drop("rn"))

# Fill missing values with a sensible default
filled = latest.withColumn("region", coalesce(col("region"), lit("UNKNOWN")))

from pyspark.sql.functions import window, avg, count

stream = (spark.readStream
    .format("delta")
    .option("readChangeFeed", "true")
    .table("bronze_events"))

agg = (stream
    .withWatermark("event_time", "10 minutes")     # tolerate 10 min of lateness
    .groupBy(window("event_time", "5 minutes"), "device_id")
    .agg(avg("temperature").alias("avg_temp"),
         count("*").alias("event_count")))

(agg.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "Files/checkpoints/sensor_agg")
    .trigger(processingTime="1 minute")
    .toTable("silver_sensor_agg"))

// Tumbling: bin() rounds each timestamp down to the window start
SensorReadings
| summarize Events = count(), AvgTemp = avg(Temperature)
    by bin(Timestamp, 5m), DeviceId

// Hopping: 15-minute window reported every 5 minutes
SensorReadings
| extend WindowStart = bin(Timestamp, 5m)
| summarize AvgTemp = avg(Temperature)
    by WindowStart, DeviceId
| order by WindowStart asc