Process with Spark Structured Streaming — Big Data & Distributed Computing

This exercise implements Stage 2 of the capstone: the Spark Structured Streaming processing layer that reads from the three IoT Kafka topics, applies watermarks for late sensor data, computes 1-minute tumbling window aggregates per sensor type and match, detects anomalies using a rolling baseline built from the windowed results, and writes anomaly alerts to a Kafka output topic. The pipeline uses `foreachBatch` for the Delta write sink and verifies the physical plan shows `BroadcastHashJoin` for the small anomaly threshold lookup table.

Analogy🏏Cricket

🏏 Think of it like cricket: Imagine the DRS review system deployed across three independent video-review centres in Mumbai, Chennai, and London, each holding a copy of the ball-tracking data. A CAP partition is a network outage that cuts communication between them. A CP system says: if the centres cannot synchronise, no review decision is issued — no player walks until the system is restored. Consistency is guaranteed; availability is sacrificed. An AP system says: each centre issues its own decision based on its local data, even if that means Mumbai says out and London says not out — reviews continue but different centres may give different verdicts. Partition tolerance is non-negotiable because the network always has the possibility of failing; the choice is whether umpires wait for consensus or proceed with local data.

The streaming pipeline reads IoT events from local Parquet files written from the Stage 1 output — the same pattern as Module 5's practice exercise. All Structured Streaming concepts from Module 5 are applied in sequence: Kafka source schema parsing, watermark definition, tumbling window aggregation, `append` output mode for the windowed results, and `trigger(once=True)` for deterministic testing. The anomaly detection uses a second-pass window function on the materialised aggregates, exactly as demonstrated in the Module 5 practice exercise.

Step 1 — Write Stage 1 Output to Parquet and Build Stream

Materialise the 300 IoT events from Stage 1 into three Parquet files — one per sensor type — with proper timestamp columns. Build the Structured Streaming pipeline that reads all three files, applies a 30-second watermark, and computes 1-minute tumbling window aggregates: mean heart rate and max acceleration per player per window for wearables, mean ball speed and impact zone distribution for pitch sensors, and mean decibels and peak density per zone for crowd sensors. Write all three aggregation streams to separate Delta table paths.

Analogy🏏Cricket

🏏 Think of it like cricket: the three per-sensor streams are three specialist analysts each watching one feed — the physio on wearables, the pitch analyst on ball impacts, the operations officer on crowd noise — each compiling their own over-by-over summary independently, which is exactly the per-sensor 1-minute window aggregation. The 30-second watermark is each analyst's rule for closing an over's figures: hold the page briefly for a delayed reading, then rule it off and move on. The three separate checkpoint locations are each analyst keeping their own bookmark of where they are up to in their own feed — and the reason they must never share one is obvious the moment you picture it: two scorers sharing a single bookmark in two different books would each keep opening to the other's page, scrambling both records. Per-query checkpoints keep each stream's offsets and state private. The payoff is three clean, independently correct summary tables, each finalised over by over, ready for the anomaly pass.

python

# capstone_spark_streaming.py — Step 1: Stream setup and windowed aggregations
from pyspark.sql import SparkSession, functions as F, Window
from pyspark.sql.types import *
from delta import configure_spark_with_delta_pip
import pandas as pd
import numpy as np
from datetime import datetime, timezone, timedelta
from pathlib import Path
import shutil

np.random.seed(42)

builder = (
    SparkSession.builder.appName("IPL-IoT-Streaming").master("local[4]")
    .config("spark.sql.shuffle.partitions",  "8")
    .config("spark.sql.adaptive.enabled",    "true")
    .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()

# Write Stage 1 events to Parquet (simulates Kafka → HDFS landing zone)
base_ts   = datetime(2024, 4, 20, 19, 30, 0, tzinfo=timezone.utc)
MATCHES   = [10001, 10002, 10003]
PLAYERS   = [(1,"Rohit"),(2,"Kohli"),(3,"Bumrah"),(4,"Dhoni")]
ZONES     = ["North","South","East","West"]
SOURCE    = Path("/tmp/iot_source")
DELTA_BASE= "/tmp/iot_delta"

if SOURCE.exists(): shutil.rmtree(SOURCE)
SOURCE.mkdir(parents=True)

wearable_rows, pitch_rows, crowd_rows = [], [], []
for i in range(100):
    ts = base_ts + timedelta(seconds=i)
    m  = np.random.choice(MATCHES)
    pid, pname = PLAYERS[i % len(PLAYERS)]
    wearable_rows.append({"event_id":i+1, "match_id":int(m),
        "player_id":pid, "player_name":pname,
        "heart_rate":int(np.random.normal(155,15)),
        "acceleration":round(float(np.random.uniform(0,8)),2),
        "running_dist":round(float(np.random.uniform(0,10)),2),
        "event_ts": ts})
    pitch_rows.append({"event_id":i+101, "match_id":int(m), "delivery_id":i+1,
        "impact_x":round(float(np.random.uniform(-1.5,1.5)),3),
        "impact_y":round(float(np.random.uniform(0,22)),3),
        "ball_speed":round(float(np.random.uniform(110,145)),1),
        "spin_rate":round(float(np.random.choice([0.0,float(np.random.uniform(1000,3000))])),0),
        "event_ts":ts})
    crowd_rows.append({"event_id":i+201, "match_id":int(m),
        "zone_id":np.random.choice(ZONES),
        "decibels":round(float(np.random.normal(85,12)),1),
        "density":round(float(np.random.uniform(0.5,4.0)),2),
        "event_ts":ts})

for name, rows in [("wearable",wearable_rows),("pitch",pitch_rows),("crowd",crowd_rows)]:
    spark.createDataFrame(pd.DataFrame(rows)).write.mode("overwrite") \
        .parquet(str(SOURCE / name))
    print(f"  {name}: {len(rows)} events written to Parquet")

# Schemas
WEARABLE_SCHEMA = StructType([
    StructField("event_id",    IntegerType(),False), StructField("match_id",IntegerType(),False),
    StructField("player_id",   IntegerType(),False), StructField("player_name",StringType(),False),
    StructField("heart_rate",  IntegerType(),False), StructField("acceleration",FloatType(),False),
    StructField("running_dist",FloatType(),  False), StructField("event_ts",TimestampType(),False),
])
PITCH_SCHEMA = StructType([
    StructField("event_id",   IntegerType(),False), StructField("match_id",IntegerType(),False),
    StructField("delivery_id",IntegerType(),False), StructField("impact_x",FloatType(),False),
    StructField("impact_y",   FloatType(),  False), StructField("ball_speed",FloatType(),False),
    StructField("spin_rate",  FloatType(),  False), StructField("event_ts",TimestampType(),False),
])
CROWD_SCHEMA = StructType([
    StructField("event_id",IntegerType(),False), StructField("match_id",IntegerType(),False),
    StructField("zone_id", StringType(), False), StructField("decibels",FloatType(),False),
    StructField("density", FloatType(),  False), StructField("event_ts",TimestampType(),False),
])

def read_stream(schema, path):
    return (spark.readStream.schema(schema).parquet(path)
            .withWatermark("event_ts", "30 seconds"))

wearable_stream = read_stream(WEARABLE_SCHEMA, str(SOURCE/"wearable"))
pitch_stream    = read_stream(PITCH_SCHEMA,    str(SOURCE/"pitch"))
crowd_stream    = read_stream(CROWD_SCHEMA,    str(SOURCE/"crowd"))

# 1-minute tumbling windows
def window_agg(stream, groupby_cols, agg_exprs, delta_path, chk_path):
    q = (stream
         .groupBy(groupby_cols + [F.window("event_ts","1 minute")])
         .agg(*agg_exprs)
         .withColumn("window_start",F.col("window.start"))
         .withColumn("window_end",  F.col("window.end")).drop("window")
         .writeStream.outputMode("append").format("delta")
         .option("checkpointLocation", chk_path)
         .trigger(once=True).start(delta_path))
    q.awaitTermination()
    print(f"  {delta_path}: {spark.read.format('delta').load(delta_path).count()} window rows")

window_agg(wearable_stream, ["match_id","player_name"],
    [F.round(F.avg("heart_rate"),1).alias("avg_hr"),
     F.round(F.max("acceleration"),2).alias("max_accel")],
    f"{DELTA_BASE}/wearable_windows", "/tmp/chk/wearable")

window_agg(pitch_stream, ["match_id"],
    [F.round(F.avg("ball_speed"),1).alias("avg_speed"),
     F.round(F.avg("spin_rate"),0).alias("avg_spin"),
     F.count("*").alias("deliveries")],
    f"{DELTA_BASE}/pitch_windows", "/tmp/chk/pitch")

window_agg(crowd_stream, ["match_id","zone_id"],
    [F.round(F.avg("decibels"),1).alias("avg_db"),
     F.round(F.max("density"),2).alias("peak_density")],
    f"{DELTA_BASE}/crowd_windows", "/tmp/chk/crowd")

print("\nStep 1 ✓: all three windowed aggregations written to Delta")

Step 2 — Anomaly Detection on Wearable Data

Read the wearable window aggregates from Delta, compute each player's rolling 3-window mean and standard deviation for heart rate, and flag windows where `avg_hr` exceeds mean + 1.5 standard deviations as fatigue alerts. Write the alert flags back to Delta via MERGE. Assert that at least one player has at least one window flagged as a fatigue alert (given the random heart rate distribution with mean 155 and SD 15), and verify the MERGE is idempotent by running it twice and asserting the alert count is unchanged.

Analogy🏏Cricket

🏏 Think of it like cricket: a physio never judges players against one universal heart-rate threshold — a fast bowler hitting 160 bpm in his fourth over of a spell may be entirely normal, while a spinner spiking to 150 between overs is a red flag. The rolling 3-window mean and standard deviation build exactly that personal recent-workload baseline, and flagging only above mean plus 1.5 standard deviations is the physio's discipline: one elevated reading is noise, a sustained departure from the player's own recent range is a fatigue signal worth sending to the dugout. Writing the flags back via an update-only MERGE — deliberately without an insert clause — is annotating the already-recorded over in the workload log rather than inventing new entries: an alert can only mark an over that was actually bowled and recorded, so the annotation pass can never fabricate a window the aggregation never produced. The payoff is an alert stream the coaching staff can act on mid-match, grounded entirely in overs that really happened.

python

# capstone_spark_streaming.py — Step 2: Anomaly detection on wearable data
from delta import DeltaTable

WEARABLE_DELTA = f"{DELTA_BASE}/wearable_windows"

# Read wearable aggregates and compute rolling baseline
wearable_df = spark.read.format("delta").load(WEARABLE_DELTA)

player_time_window = (
    Window.partitionBy("player_name")
    .orderBy("window_start")
    .rowsBetween(-2, 0)  # 3-window rolling
)

with_stats = (
    wearable_df
    .withColumn("rolling_hr_mean",
        F.round(F.avg("avg_hr").over(player_time_window), 1))
    .withColumn("rolling_hr_std",
        F.round(F.stddev("avg_hr").over(player_time_window), 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)
    )
)

alerts = with_stats.filter(F.col("is_fatigue_alert") == True)
print(f"\nFatigue alerts detected: {alerts.count()}")
print("Alert details:")
alerts.select("player_name","match_id","window_start",
              "avg_hr","rolling_hr_mean","rolling_hr_std").show(truncate=False)

assert alerts.count() > 0, "Expected at least one fatigue alert given heart rate distribution"

# MERGE alert flags back to Delta
delta_tbl = DeltaTable.forPath(spark, WEARABLE_DELTA)

def merge_alerts(df_with_flags):
    delta_tbl.alias("t").merge(
        df_with_flags.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()

merge_alerts(with_stats)
count_1 = spark.read.format("delta").load(WEARABLE_DELTA) \
    .filter(F.col("is_fatigue_alert") == True).count()

# Idempotency: second MERGE must not change alert count
merge_alerts(with_stats)
count_2 = spark.read.format("delta").load(WEARABLE_DELTA) \
    .filter(F.col("is_fatigue_alert") == True).count()

assert count_1 == count_2, f"Idempotency failed: {count_1} → {count_2} alerts"
print(f"\nIdempotency verified: {count_1} alerts unchanged after re-MERGE ✓")
print("Step 2 complete: anomaly detection pipeline verified")

Warning: `F.stddev` returns `null` for groups with only one data point (cannot compute standard deviation from a single observation). The `F.coalesce(F.col('rolling_hr_std'), F.lit(0.0))` in the anomaly condition handles this by treating single-window baselines as having zero standard deviation — meaning the anomaly threshold equals the rolling mean exactly, flagging any window that exceeds its own mean. In production, set a minimum window count threshold (e.g. require at least 3 windows before flagging anomalies) to avoid false positives from single-observation baselines.

Extension Challenge: Add a third step that joins the wearable fatigue alerts with the pitch window aggregates on `match_id` and `window_start` to correlate player fatigue with pitch deterioration. Use a broadcast join on the pitch aggregates (fewer rows) and identify windows where both fatigue alerts and above-average ball spin rates occurred simultaneously — the hypothesis being that spin bowlers are more effective when batters are fatigued. Write the correlated results to a third Delta table `ipl-fatigue-pitch-correlation`.