This exercise builds a complete Spark batch job that processes simulated IPL match clickstream data — fan engagement events from a mobile app recorded during live matches. The pipeline reads raw event Parquet files from HDFS, applies schema validation, enriches events with match metadata via a broadcast join, computes per-match and per-bowler engagement statistics using window functions, writes results to a Delta table with upsert semantics, and demonstrates partition-count tuning and skew detection. Every concept from Module 2 Lessons 1–5 is exercised in sequence.
The clickstream dataset models real-world event log characteristics: high volume, heterogeneous event types, a heavily skewed user distribution where a small number of power users generate a disproportionate share of events, and a match_id foreign key that joins to a small metadata table. These characteristics make it a realistic proxy for production Spark workloads — the skew, the broadcast join opportunity, and the Delta upsert requirement all arise naturally from the domain rather than being artificially constructed.
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 — Data Generation and Schema Validation
Generate the synthetic clickstream dataset as a Parquet file with intentional schema characteristics: a heavy-tailed user distribution, event type variety, and a skewed match distribution where one match receives five times the normal event count. Apply explicit schema validation using `df.schema` comparison after reading, asserting all required columns are present and correctly typed before any transformation logic runs.
Analogy🏏Cricket
🏏 Think of it like cricket: validating the schema before any transformation is the match referee's team-sheet check before the toss — confirm every column is present and correctly typed before a single ball is bowled, because discovering a missing name in the 15th over voids everything played since. The dataset's deliberate quirks mirror a real season: the heavy-tailed user distribution is the superfan phenomenon — a handful of die-hards who react to every ball generate a huge share of all noise, just as a few power users dominate an event log. The skewed match with five times the events is the blockbuster final: one fixture draws a crowd no ordinary league game approaches, and any pipeline that assumes matches are equal will choke on it. Just as a good analyst checks the fixture list and knows which match will strain the systems, generating the skew deliberately and validating the schema up front means every later step is built for the data you actually have, not the data you wish you had.
python
# exercise_spark_batch.py — Step 1: Data generation and schema validation
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
StructType, StructField, LongType, IntegerType,
StringType, TimestampType, BooleanType
)
import random
from datetime import datetime, timezone, timedelta
import pandas as pd
import numpy as np
random.seed(42)
np.random.seed(42)
spark = (
SparkSession.builder
.appName("IPL-Clickstream-Batch")
.master("local[4]")
.config("spark.sql.adaptive.enabled", "true")
.config("spark.sql.shuffle.partitions", "50")
.config("spark.sql.extensions",
"io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.delta.catalog.DeltaCatalog")
.getOrCreate()
)
# Generate synthetic clickstream data
MATCH_IDS = [10001, 10002, 10003, 10004]
EVENT_TYPES= ["boundary_reaction","wicket_alert","score_refresh",
"live_chat","prediction_submit"]
N_EVENTS = 5000
# Heavy-tailed: match 10001 gets 5x normal events (skew)
match_weights = [0.50, 0.20, 0.15, 0.15] # 10001 is the final — huge skew
base_ts = datetime(2024, 4, 20, 19, 30, 0, tzinfo=timezone.utc)
events = pd.DataFrame({
"event_id": range(1, N_EVENTS+1),
"match_id": np.random.choice(MATCH_IDS, N_EVENTS, p=match_weights),
"user_id": np.random.zipf(1.5, N_EVENTS).clip(1, 500), # zipf = power-law
"event_type": np.random.choice(EVENT_TYPES, N_EVENTS),
"over": np.random.randint(1, 21, N_EVENTS),
"ts": [base_ts + timedelta(seconds=int(s))
for s in np.random.uniform(0, 7200, N_EVENTS)],
"is_premium": np.random.choice([True, False], N_EVENTS, p=[0.2, 0.8]),
})
# Write to local Parquet (simulates HDFS)
events_path = "/tmp/ipl_clickstream/events.parquet"
spark.createDataFrame(events).write.mode("overwrite").parquet(events_path)
print(f"Generated {N_EVENTS:,} events across {len(MATCH_IDS)} matches")
print(f"Match distribution: {dict(zip(MATCH_IDS, np.unique(events['match_id'], return_counts=True)[1]))}")
# Define and validate schema
EXPECTED_SCHEMA = StructType([
StructField("event_id", LongType(), False),
StructField("match_id", LongType(), False),
StructField("user_id", LongType(), False),
StructField("event_type", StringType(), False),
StructField("over", LongType(), False),
StructField("ts", TimestampType(), False),
StructField("is_premium", BooleanType(), False),
])
events_df = spark.read.schema(EXPECTED_SCHEMA).parquet(events_path)
required_cols = {f.name for f in EXPECTED_SCHEMA.fields}
actual_cols = {f.name for f in events_df.schema.fields}
assert required_cols.issubset(actual_cols), f"Missing columns: {required_cols - actual_cols}"
print(f"Schema validated: {len(actual_cols)} columns, {events_df.count():,} rows")Step 2 — Broadcast Join, Enrichment and Feature Engineering
Create the small match metadata DataFrame (four rows), force a broadcast join onto the events DataFrame to eliminate the shuffle, add derived columns for `phase`, `hour_of_day`, and `is_boundary_event`, and verify the physical plan confirms `BroadcastHashJoin` rather than `SortMergeJoin`. Assert that the enriched DataFrame has the same row count as the original events, confirming no rows were lost in the left join.
Analogy🏏Cricket
🏏 Think of it like cricket: the broadcast join is how a sensible statistician attaches venue details to millions of delivery records. The wrong way is to ship every delivery slip to a central office to be matched against the fixture list — a colossal movement of paper for a four-line lookup. The right way is to photocopy the one-page fixture card and hand a copy to every scorer, who annotates their own records locally: that is exactly what broadcasting the four-row metadata table does, copying it to every executor so no event ever travels across the network. Checking the physical plan for BroadcastHashJoin is the captain confirming the field is actually set the way he signalled — intent is not execution. The derived columns are the enrichments every analyst adds: which phase of the innings, what hour, was it a boundary moment. And asserting the row count is unchanged is the golden rule of scoring: enrichment may annotate deliveries, but not one ball may go missing from the book.
python
# exercise_spark_batch.py — Step 2: Broadcast join and feature engineering
# Match metadata — small table, perfect candidate for broadcast
match_meta = spark.createDataFrame([
(10001, "Wankhede Stadium", "Mumbai Indians", "CSK", "final"),
(10002, "Chinnaswamy Stadium","RCB", "KKR", "qualifier"),
(10003, "Eden Gardens", "KKR", "SRH", "league"),
(10004, "Chepauk", "CSK", "RCB", "league"),
], ["match_id", "venue", "home_team", "away_team", "match_type"])
# Broadcast join — eliminates shuffle, verified in physical plan
enriched = (
events_df
.join(F.broadcast(match_meta), on="match_id", how="left")
.withColumn("phase",
F.when(F.col("over") <= 6, "powerplay")
.when(F.col("over") <= 15, "middle")
.otherwise("death")
)
.withColumn("hour_of_day", F.hour(F.col("ts")))
.withColumn("is_boundary_event",
F.col("event_type").isin(["boundary_reaction"]))
.withColumn("is_wicket_event",
F.col("event_type") == "wicket_alert")
)
# Verify broadcast join in plan
print("Physical plan — should show BroadcastHashJoin:")
enriched.explain("formatted")
# Assert no rows lost
assert enriched.count() == N_EVENTS, "Row count changed after broadcast join"
print(f"\nEnriched: {enriched.count():,} rows — row count preserved ✓")
# Show skew in partition distribution
partition_dist = (
enriched
.withColumn("pid", F.spark_partition_id())
.groupBy("pid").count()
.orderBy("count", ascending=False)
)
print("\nPartition row count distribution (top 5):")
partition_dist.show(5)Step 3 — Window Functions, Aggregation and Delta Write
Compute engagement statistics at two levels: per-match summary (total events, unique users, peak hourly engagement) and per-phase breakdown (events and premium-user fraction per phase per match). Use window functions to rank matches by engagement. Write the per-match summary to a Delta table with MERGE semantics. Verify the Delta time travel feature by querying the table's version 0 after a second write updates two records, confirming the original values are still accessible.
Analogy🏏Cricket
🏏 Think of it like cricket: the two aggregation levels are the two tables every league publishes: the match summary (total attendance, engagement, peak moments per fixture) and the phase breakdown — powerplay, middle overs, death — showing where within each match the excitement lived. Ranking matches by engagement is like compiling an orange-cap-style leaderboard: a rank is only meaningful once every candidate's total sits side by side, which is why the window function must see all rows together. The Delta MERGE is the master season table maintained properly — when a match's corrected figures arrive, you update that one line in place rather than tearing up the whole sheet and rewriting it, and re-running the update never double-counts. Time travel to version 0 is the archived carbon copy: you can always re-read the table exactly as it stood after the first compilation. The payoff is a season record you can correct, audit, and rank without ever corrupting what was already published.
python
# exercise_spark_batch.py — Step 3: Aggregation, window functions, Delta write
from pyspark.sql import Window
from delta import DeltaTable
DELTA_PATH = "/tmp/ipl_match_engagement/"
# --- Per-match engagement summary ---
match_summary = (
enriched
.groupBy("match_id", "venue", "match_type")
.agg(
F.count("event_id").alias("total_events"),
F.countDistinct("user_id").alias("unique_users"),
F.sum(F.col("is_premium").cast("int")).alias("premium_events"),
F.sum(F.col("is_boundary_event").cast("int")).alias("boundary_reactions"),
F.sum(F.col("is_wicket_event").cast("int")).alias("wicket_alerts"),
)
.withColumn("premium_pct",
F.round(F.col("premium_events") / F.col("total_events") * 100, 1))
)
# Window function: rank matches by total engagement
window_spec = Window.orderBy(F.col("total_events").desc())
match_ranked = match_summary.withColumn(
"engagement_rank", F.rank().over(window_spec)
)
print("Match engagement ranking:")
match_ranked.orderBy("engagement_rank").show(truncate=False)
# --- Per-phase breakdown ---
phase_stats = (
enriched
.groupBy("match_id", "phase")
.agg(
F.count("event_id").alias("events"),
F.countDistinct("user_id").alias("unique_users"),
F.round(F.avg(F.col("is_premium").cast("double")), 3).alias("premium_rate"),
)
.orderBy("match_id", "phase")
)
print("\nPer-phase engagement stats:")
phase_stats.show(truncate=False)
# --- Write to Delta with MERGE ---
match_ranked.select(
"match_id","venue","match_type","total_events",
"unique_users","premium_pct","engagement_rank"
).write.format("delta").mode("overwrite").save(DELTA_PATH)
print("\nInitial Delta write complete (version 0)")
# Simulate an update: match 10001 engagement revised after re-processing
updated = spark.createDataFrame([
(10001, "Wankhede Stadium", "final", 2650, 390, 21.5, 1),
], ["match_id","venue","match_type","total_events","unique_users","premium_pct","engagement_rank"])
delta_tbl = DeltaTable.forPath(spark, DELTA_PATH)
delta_tbl.alias("t").merge(
updated.alias("s"), "t.match_id = s.match_id"
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
print("MERGE applied — version 1 committed")
# Time travel: confirm version 0 still shows original value
v0 = spark.read.format("delta").option("versionAsOf", 0).load(DELTA_PATH)
v0_total = v0.filter(F.col("match_id") == 10001).select("total_events").collect()[0][0]
v1 = spark.read.format("delta").load(DELTA_PATH)
v1_total = v1.filter(F.col("match_id") == 10001).select("total_events").collect()[0][0]
print(f"\nTime travel verification:")
print(f" Version 0 match_id=10001 total_events: {v0_total}")
print(f" Version 1 match_id=10001 total_events: {v1_total}")
assert v1_total == 2650, f"MERGE did not update total_events: {v1_total}"
assert v0_total != v1_total, "Time travel failed: versions should differ"
print("All assertions passed. Spark batch job complete.")