Practice — Real-Time Anomaly Detection Pipeline — Big Data & Distributed Computing

This exercise builds a real-time anomaly detection pipeline for IPL match delivery events using Spark Structured Streaming with all five Module 5 concepts: reading from a simulated Kafka source, applying watermarks for late-data handling, computing tumbling window statistics, detecting statistical anomalies using a custom threshold, writing results to Delta Lake with idempotent MERGE semantics, and verifying exactly-once behaviour by replaying the same batch and asserting zero row count change in the output table.

The anomaly detection logic flags deliveries where a bowler's economy in the current over window exceeds two standard deviations above their rolling 3-window baseline. This is the type of real-time alerting that batch pipelines cannot deliver — by the time the daily batch runs, the match is over and the insight has no operational value. The exercise uses a local file-based streaming source rather than a live Kafka cluster, but the transformation, windowing, and sink logic is identical to a production Kafka-backed pipeline.

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.

Step 1 — Streaming Source and Windowed Economy

Generate a synthetic delivery dataset with known anomalies — two overs where a specific bowler's economy is artificially elevated — and write it as a streaming Parquet source. Build the Structured Streaming pipeline that reads from this source, applies a 2-minute watermark, computes tumbling 5-minute windowed economy per bowler, and writes the window results to a Delta table. Verify the window results DataFrame has the expected column schema before proceeding to anomaly detection.

Analogy🏏Cricket

🏏 Think of it like cricket: the tumbling window is scoring the game the way cricket already scores itself: in overs. Each 5-minute window, like an over, is a fixed, non-overlapping unit — it opens, collects its deliveries, closes, and the next one begins; a delivery belongs to exactly one over and a reading to exactly one window. Computing windowed economy per bowler is keeping each bowler's over-by-over figures rather than waiting for close of play. The 2-minute watermark is the third umpire's replay deadline: when an over ends, the officials wait a bounded time for any delayed feed from the broadcast truck before finalising that over's figures — a signal that arrives after the deadline is noted but cannot reopen a sealed over, because a scoreboard that never finalises anything is useless to a captain making live decisions. Writing each closed window to Delta is entering the completed over in the official book. The payoff is the essence of streaming: figures the dugout can act on while the match is still being played.

python

# exercise_streaming_anomaly.py — Step 1: Streaming source and windowed economy
from pyspark.sql import SparkSession, functions as F
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-Anomaly-Detection").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()

# ── Generate synthetic delivery stream with injected anomalies ────────────────
BOWLERS    = ["Bumrah", "Shami", "Hardik"]
base_ts    = datetime(2024, 4, 20, 19, 30, 0, tzinfo=timezone.utc)
records    = []
del_id     = 1

for over in range(1, 9):       # 8 overs
    for bowler in BOWLERS:
        for ball in range(1, 7):
            ts = base_ts + timedelta(minutes=(over-1)*5, seconds=ball*8)
            # Inject anomaly: Bumrah concedes 6 runs every ball in overs 5 and 7
            if bowler == "Bumrah" and over in (5, 7):
                runs = 6
            else:
                runs = int(np.random.choice([0,1,2,4,6], p=[0.35,0.30,0.15,0.15,0.05]))
            records.append({
                "delivery_id": del_id,
                "match_id":    10001,
                "over":        over,
                "ball":        ball,
                "bowler":      bowler,
                "runs":        runs,
                "is_wicket":   bool(np.random.random() < 0.05),
                "event_ts":    ts,
            })
            del_id += 1

STREAM_PATH = Path("/tmp/ipl_stream_source")
DELTA_PATH  = "/tmp/ipl_window_stats"
CHECKPT     = "/tmp/ipl_checkpoints/anomaly_step1"

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

# Write as Parquet files (simulates Kafka micro-batches landing on HDFS)
df_source = pd.DataFrame(records)
spark.createDataFrame(df_source).write.mode("overwrite").parquet(str(STREAM_PATH))
print(f"Generated {len(records)} delivery records with anomalies in overs 5 and 7")

# Define schema
DELIVERY_SCHEMA = StructType([
    StructField("delivery_id", IntegerType(),  False),
    StructField("match_id",    IntegerType(),  False),
    StructField("over",        IntegerType(),  False),
    StructField("ball",        IntegerType(),  False),
    StructField("bowler",      StringType(),   False),
    StructField("runs",        IntegerType(),  False),
    StructField("is_wicket",   BooleanType(),  False),
    StructField("event_ts",    TimestampType(),False),
])

# Read as streaming DataFrame
deliveries_stream = (
    spark.readStream
    .schema(DELIVERY_SCHEMA)
    .parquet(str(STREAM_PATH))
)

# Apply watermark and compute 5-minute tumbling windowed economy
window_stats = (
    deliveries_stream
    .withWatermark("event_ts", "2 minutes")
    .groupBy(
        "bowler",
        F.window("event_ts", "5 minutes")
    )
    .agg(
        F.sum("runs").alias("window_runs"),
        F.count("*").alias("window_deliveries"),
    )
    .withColumn("window_economy",
        F.round(F.col("window_runs") / (F.col("window_deliveries") / 6.0), 2))
    .withColumn("window_start", F.col("window.start"))
    .withColumn("window_end",   F.col("window.end"))
    .drop("window")
)

# Write to Delta — append mode (watermarked aggregation)
step1_query = (
    window_stats
    .writeStream
    .outputMode("append")
    .format("delta")
    .option("checkpointLocation", CHECKPT)
    .trigger(once=True)   # process all available data, then stop
    .start(DELTA_PATH)
)
step1_query.awaitTermination()

window_df = spark.read.format("delta").load(DELTA_PATH)
print(f"\nStep 1 ✓: {window_df.count()} window records written to Delta")
expected_cols = {"bowler","window_runs","window_deliveries","window_economy","window_start","window_end"}
assert expected_cols.issubset(set(window_df.columns)), f"Missing columns"
window_df.orderBy("bowler","window_start").show(truncate=False)

Step 2 — Anomaly Detection and Delta MERGE

Read the windowed economy Delta table, compute each bowler's rolling 3-window mean and standard deviation using a Spark window function, flag windows where the economy exceeds mean + 2 standard deviations as anomalies, and write the anomaly flags back to the Delta table via MERGE. Verify that exactly the two injected anomaly windows (Bumrah's overs 5 and 7) are flagged, and confirm the MERGE is idempotent by running it twice and asserting the anomaly count does not change.

Analogy🏏Cricket

🏏 Think of it like cricket: a good bowling coach never asks 'is 12 runs an over bad?' in the abstract — he asks 'is it bad for this bowler, given his recent spells?' The rolling 3-window mean and standard deviation build exactly that personal baseline: Bumrah conceding 12 when his recent overs average 6 is an alarm, while the same 12 from a part-timer in the death overs is Tuesday. Flagging only beyond mean plus two standard deviations is the coach's discipline against overreacting — one expensive ball is noise; a spell far outside the bowler's own established range is signal worth sending to the captain. Verifying that exactly the two injected anomalies are caught is a DRS calibration test: feed the system deliveries with known edges and confirm it flags precisely those, no more, no fewer. And the idempotent MERGE means reviewing the same overs twice never double-flags them — like re-checking a scorebook never adds a second wicket. The payoff is an alert the captain can trust mid-match.

python

# exercise_streaming_anomaly.py — Step 2: Anomaly detection and Delta MERGE
from pyspark.sql import Window
from delta import DeltaTable

# Read the windowed stats written in Step 1
window_df = spark.read.format("delta").load(DELTA_PATH)

# Rolling 3-window stats per bowler using window function
bowler_window = (
    Window.partitionBy("bowler")
    .orderBy("window_start")
    .rowsBetween(-2, 0)   # current + 2 preceding = 3-window rolling
)

with_stats = (
    window_df
    .withColumn("rolling_mean",
        F.round(F.avg("window_economy").over(bowler_window), 2))
    .withColumn("rolling_std",
        F.round(F.stddev("window_economy").over(bowler_window), 3))
    .withColumn("is_anomaly",
        F.when(
            F.col("window_economy") > (
                F.col("rolling_mean") + 2 * F.coalesce(F.col("rolling_std"), F.lit(0.0))
            ), True
        ).otherwise(False)
    )
)

# Check: how many anomalies detected?
anomalies = with_stats.filter(F.col("is_anomaly") == True)
print(f"\nAnomalies detected: {anomalies.count()}")
print("Anomalous windows:")
anomalies.select("bowler","window_start","window_economy",
                 "rolling_mean","rolling_std").show(truncate=False)

# Verify the two injected anomalies are detected (Bumrah overs 5 and 7)
bumrah_anomalies = anomalies.filter(F.col("bowler") == "Bumrah")
assert bumrah_anomalies.count() >= 2, \
    f"Expected >=2 Bumrah anomalies, got {bumrah_anomalies.count()}"

# MERGE anomaly flags back to the Delta table — exactly-once idempotent
delta_tbl = DeltaTable.forPath(spark, DELTA_PATH)

delta_tbl.alias("target").merge(
    with_stats.select(
        "bowler","window_start","is_anomaly","rolling_mean","rolling_std"
    ).alias("source"),
    "target.bowler = source.bowler AND target.window_start = source.window_start"
).whenMatchedUpdate(set_={
    "is_anomaly":    "source.is_anomaly",
    "rolling_mean":  "source.rolling_mean",
    "rolling_std":   "source.rolling_std",
}).execute()
print("MERGE complete: anomaly flags written to Delta")

# Idempotency check: run MERGE again — anomaly count must not change
count_before = spark.read.format("delta").load(DELTA_PATH) \
    .filter(F.col("is_anomaly") == True).count()

delta_tbl.alias("target").merge(
    with_stats.select(
        "bowler","window_start","is_anomaly","rolling_mean","rolling_std"
    ).alias("source"),
    "target.bowler = source.bowler AND target.window_start = source.window_start"
).whenMatchedUpdate(set_={
    "is_anomaly":   "source.is_anomaly",
    "rolling_mean": "source.rolling_mean",
    "rolling_std":  "source.rolling_std",
}).execute()

count_after = spark.read.format("delta").load(DELTA_PATH) \
    .filter(F.col("is_anomaly") == True).count()

assert count_before == count_after, \
    f"Idempotency failed: {count_before} → {count_after} anomalies"
print(f"\nIdempotency verified: {count_before} anomalies before re-MERGE, {count_after} after ✓")
print("All assertions passed. Real-time anomaly detection pipeline complete.")

Warning: The `trigger(once=True)` mode used in Step 1 processes all available data in one pass and then terminates the streaming query — it is the correct mode for testing streaming pipelines against static data without requiring a live source. In production with a live Kafka source, replace `trigger(once=True)` with `trigger(processingTime='N seconds')` for continuous micro-batch processing, or `trigger(availableNow=True)` (Spark 3.3+) for incremental batch-style processing that processes all available data and stops, respecting the checkpoint for safe incremental re-runs.

Extension Challenge: Extend the pipeline to publish anomaly alerts to a Kafka topic `ipl-anomaly-alerts` using `foreachBatch` with a Kafka producer. Each alert message should include the bowler name, window start and end, current economy, rolling mean, and the number of standard deviations above baseline. Use Avro serialisation with a Schema Registry-registered schema. Verify that the number of messages in the `ipl-anomaly-alerts` topic equals the number of anomaly rows in the Delta table after the pipeline runs.