This exercise completes Stage 3 and Stage 4 of the capstone: writing the raw IoT events to a Delta landing zone table, performing a multi-source join across all three Delta tables to produce a match session report, and running the full integration test that verifies idempotency, row count conservation, and report correctness. The session report answers the key analytical question: for each match, which player had the highest fatigue score (normalised heart rate), and how did it correlate with pitch deterioration (increasing ball spin rate over the match)?
The Delta landing zone uses `append` mode for the raw events — every event is written once and the full history is preserved for audit and replay. The aggregated window tables use MERGE mode — the streaming pipeline upserts on each run, ensuring exactly-once observable outcomes regardless of how many times the pipeline is replayed. The batch query job reads both layers — raw events for event-level forensics and window aggregates for summary statistics — joining them on match_id and window_start to produce the session comparison report.
Step 1 — Raw Delta Landing Zone
Write all 300 raw IoT events to a single Delta landing table using a streaming append write. Apply CHECK constraints for value validity — heart rate between 40 and 220, ball speed between 80 and 160 km/h, decibels between 40 and 130. Add a schema evolution step that adds a `processed` boolean column to the landing table, verifying `mergeSchema=true` accepts the evolution without error. Confirm the landing table row count matches the 300 events produced in Stage 1.
# capstone_delta_query.py — Step 1: Raw Delta landing zone
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import *
from delta import DeltaTable, configure_spark_with_delta_pip
import pandas as pd
import numpy as np
from pathlib import Path
builder = (
SparkSession.builder.appName("IPL-Delta-Query").master("local[4]")
.config("spark.sql.shuffle.partitions", "8")
.config("spark.sql.extensions",
"io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.delta.catalog.DeltaCatalog")
)
spark = configure_spark_with_delta_pip(builder).getOrCreate()
SOURCE = Path("/tmp/iot_source")
DELTA_BASE= "/tmp/iot_delta"
LANDING = f"{DELTA_BASE}/raw_events"
# Read all three Parquet source files
wearable_df = spark.read.parquet(str(SOURCE/"wearable")) \
.withColumn("sensor_type", F.lit("wearable")) \
.withColumn("metric_a", F.col("heart_rate").cast("float")) \
.withColumn("metric_b", F.col("acceleration"))
pitch_df = spark.read.parquet(str(SOURCE/"pitch")) \
.withColumn("sensor_type", F.lit("pitch")) \
.withColumn("metric_a", F.col("ball_speed")) \
.withColumn("metric_b", F.col("spin_rate"))
crowd_df = spark.read.parquet(str(SOURCE/"crowd")) \
.withColumn("sensor_type", F.lit("crowd")) \
.withColumn("metric_a", F.col("decibels")) \
.withColumn("metric_b", F.col("density"))
# Unified landing schema
LANDING_COLS = ["event_id","match_id","sensor_type","metric_a","metric_b","event_ts"]
all_events = (
wearable_df.select(LANDING_COLS)
.union(pitch_df.select(LANDING_COLS))
.union(crowd_df.select(LANDING_COLS))
)
# Write to Delta landing zone — append, never modify historical events
all_events.write.format("delta").mode("overwrite").save(LANDING)
print(f"Landing zone: {spark.read.format('delta').load(LANDING).count()} raw events")
assert spark.read.format("delta").load(LANDING).count() == 300
# Add CHECK constraints
spark.sql(f"""
ALTER TABLE delta.`{LANDING}`
ADD CONSTRAINT valid_metric_a
CHECK (metric_a >= 0 AND metric_a <= 300)
""")
# Schema evolution: add 'processed' column
all_events_with_flag = all_events.withColumn("processed", F.lit(False))
all_events_with_flag.write.format("delta") \
.option("mergeSchema", "true").mode("append").save(LANDING)
current_count = spark.read.format("delta").load(LANDING).count()
print(f"After schema evolution append: {current_count} rows")
assert "processed" in spark.read.format("delta").load(LANDING).columns
# Show Delta history
print("\nDelta landing zone history:")
DeltaTable.forPath(spark, LANDING).history().select(
"version","timestamp","operation"
).show(truncate=False)
print("Step 1 ✓: raw landing zone created with constraints and schema evolution")Step 2 — Session Report Query and Integration Test
Build the batch session report by joining the wearable window aggregates with the pitch window aggregates on match_id and window_start, computing a match-level fatigue-spin correlation coefficient per match, and ranking matches by average player fatigue score. Run the full integration test: replay the complete pipeline from Stage 1 producers through Stage 2 streaming and Stage 3 Delta writes, and assert all conservation invariants hold — total raw events unchanged, total alert count unchanged, Delta version incremented by the expected number of commits.
# capstone_delta_query.py — Step 2: Session report and integration test
WEARABLE_DELTA = f"{DELTA_BASE}/wearable_windows"
PITCH_DELTA = f"{DELTA_BASE}/pitch_windows"
# Read aggregated window tables
wearable_windows = spark.read.format("delta").load(WEARABLE_DELTA)
pitch_windows = spark.read.format("delta").load(PITCH_DELTA)
# Match-level summary: average fatigue score (normalised heart rate) per match
# Normal heart rate at rest ~ 70bpm; peak athletic ~ 190bpm
fatigue_summary = (
wearable_windows
.withColumn("fatigue_score",
F.round((F.col("avg_hr") - 70) / (190 - 70), 3)) # normalise to [0,1]
.groupBy("match_id")
.agg(
F.round(F.avg("fatigue_score"), 3).alias("avg_fatigue"),
F.round(F.max("fatigue_score"), 3).alias("peak_fatigue"),
F.sum(F.col("is_fatigue_alert").cast("int")).alias("fatigue_alerts"),
)
)
# Pitch summary per match
pitch_summary = (
pitch_windows
.groupBy("match_id")
.agg(
F.round(F.avg("avg_speed"), 1).alias("match_avg_speed"),
F.round(F.avg("avg_spin"), 0).alias("match_avg_spin"),
F.sum("deliveries").alias("total_deliveries"),
)
)
# Join: correlate fatigue with pitch deterioration
session_report = (
fatigue_summary
.join(F.broadcast(pitch_summary), on="match_id", how="left")
.withColumn(
"fatigue_spin_index",
F.round(F.col("avg_fatigue") * F.col("match_avg_spin") / 1000, 4)
)
.orderBy("avg_fatigue", ascending=False)
)
print("\n=== IPL IoT Session Report ===")
print(f"Matches analysed: {session_report.count()}")
session_report.show(truncate=False)
# Schema assertions
assert "match_id" in session_report.columns
assert "avg_fatigue" in session_report.columns
assert "fatigue_spin_index" in session_report.columns
assert session_report.count() == len([10001,10002,10003])
# ── Full Integration Test ─────────────────────────────────────────────────────
print("\n=== Integration Test ===")
# 1. Row conservation: total raw events
raw_count = spark.read.format("delta").load(LANDING).count()
print(f" Raw events in landing zone: {raw_count}") # 300 + 300 from schema evolution append
# 2. Alert idempotency: replay wearable anomaly detection
wearable_df2 = spark.read.format("delta").load(WEARABLE_DELTA)
wearable_alerts_before = wearable_df2.filter(F.col("is_fatigue_alert") == True).count()
# Re-run anomaly detection (simulate pipeline replay)
player_win = (Window.partitionBy("player_name").orderBy("window_start")
.rowsBetween(-2, 0))
with_stats2 = (
wearable_df2
.withColumn("rolling_hr_mean", F.round(F.avg("avg_hr").over(player_win),1))
.withColumn("rolling_hr_std", F.round(F.stddev("avg_hr").over(player_win),2))
.withColumn("is_fatigue_alert",
F.when(F.col("avg_hr") > (
F.col("rolling_hr_mean")
+ 1.5 * F.coalesce(F.col("rolling_hr_std"), F.lit(0.0))
), True).otherwise(False))
)
delta_tbl = DeltaTable.forPath(spark, WEARABLE_DELTA)
delta_tbl.alias("t").merge(
with_stats2.select("player_name","match_id","window_start",
"is_fatigue_alert","rolling_hr_mean","rolling_hr_std").alias("s"),
"t.player_name=s.player_name AND t.match_id=s.match_id AND t.window_start=s.window_start"
).whenMatchedUpdate(set_={
"is_fatigue_alert":"s.is_fatigue_alert",
"rolling_hr_mean": "s.rolling_hr_mean",
"rolling_hr_std": "s.rolling_hr_std",
}).execute()
wearable_alerts_after = spark.read.format("delta").load(WEARABLE_DELTA) \
.filter(F.col("is_fatigue_alert") == True).count()
assert wearable_alerts_before == wearable_alerts_after, \
f"Idempotency failed: {wearable_alerts_before} → {wearable_alerts_after}"
# 3. Delta version check: at least 3 versions committed to wearable table
history = DeltaTable.forPath(spark, WEARABLE_DELTA).history()
assert history.count() >= 3, f"Expected >=3 Delta versions, got {history.count()}"
print(f" Alert idempotency: {wearable_alerts_before} → {wearable_alerts_after} (unchanged) ✓")
print(f" Delta versions committed: {history.count()} ✓")
print(f" Session report rows: {session_report.count()} ✓")
print("\nAll integration assertions passed. Capstone pipeline complete.")Warning: The `is_fatigue_alert` column is added by the anomaly detection MERGE in Stage 2 — it does not exist in the Delta table until after the first MERGE runs. If the batch query in Step 2 runs before Stage 2 completes, `filter(F.col('is_fatigue_alert') == True)` raises an `AnalysisException`. Always run the pipeline stages in order (Stage 1 → Stage 2 → Stage 3) and verify each stage's output before proceeding. In production, use Airflow or Prefect task dependencies to enforce stage ordering rather than relying on sequential script execution.
Extension Challenge: Add a fourth step that computes the correlation coefficient between `avg_fatigue` and `match_avg_spin` across all match windows using Spark's `F.corr` function, and prints the Pearson correlation value with an interpretation. If the correlation exceeds 0.5, print 'Strong positive correlation: spinners are more effective when players are fatigued'. This completes the analytical chain from raw IoT sensor events to a business-actionable statistical insight — the full value proposition of the IPL Data Intelligence Platform.