This exercise builds a complete PySpark feature engineering pipeline for IPL match outcome prediction. You will implement window functions for rolling player statistics, join multiple source DataFrames including a broadcast join, apply a Pandas UDF for custom normalisation, define and enforce a Delta Lake schema with CHECK constraints, write the feature set to a Delta table with MERGE semantics, and verify the physical plan to confirm expected join strategies. The pipeline produces a versioned, auditable feature dataset ready for consumption by a downstream ML training job.
The exercise is structured in four steps. Step 1 generates source data and constructs the window-function-based rolling player statistics. Step 2 joins all feature sources, applies the Pandas UDF normalisation, and validates the schema. Step 3 writes to Delta with constraints and verifies the transaction log. Step 4 reads the feature set back, applies the MLlib preprocessing pipeline, and confirms the end-to-end feature vector output is correct before the pipeline is considered complete.
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 — Rolling Window Statistics
Generate five matches of delivery data and compute three rolling features for each bowler per match: rolling 3-match average economy, rolling 3-match wicket rate, and overall season cumulative wickets at the point of each match. These rolling statistics capture form context that single-match statistics miss, which is why they are the most predictive features in short-form cricket outcome models. Verify that all three window functions produce finite values for all rows by asserting zero nulls in the output columns.
Analogy🏏Cricket
🏏 Think of it like cricket: no selector picks a squad from single-match figures — one bad over on a green pitch means nothing; the last three matches mean everything. The rolling 3-match economy and wicket rate are exactly a selector's form guide: for each bowler, at each match, look back over the most recent three outings and average them, so the number travels with the player through the season rather than freezing at one snapshot. The cumulative season wickets column is the running tally printed next to each match in the almanac — 'as of match 4, Bumrah had 11 wickets' — computed from the season's start up to that point, never peeking ahead. Just as a form guide is worthless if the matches are listed out of order, the window computation depends on partitioning by bowler and ordering by match before scanning. The payoff is precisely why these are the most predictive features: the model, like a good captain choosing his death bowler, gets to judge current form, not just career reputation.
python
# exercise_pyspark_features.py — Step 1: Rolling window statistics
from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F
from pyspark.sql.types import *
import numpy as np
np.random.seed(42)
spark = (
SparkSession.builder.appName("IPL-Features").master("local[4]")
.config("spark.sql.adaptive.enabled", "true")
.config("spark.sql.shuffle.partitions", "50")
.config("spark.sql.execution.arrow.pyspark.enabled", "true")
.config("spark.sql.extensions",
"io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.delta.catalog.DeltaCatalog")
.getOrCreate()
)
BOWLERS = ["Bumrah", "Shami", "Ashwin", "Jadeja", "Hardik"]
MATCH_IDS = list(range(10001, 10006)) # 5 matches
# Generate delivery data
deliveries_rows = []
for mid in MATCH_IDS:
for bowler in BOWLERS:
n_balls = np.random.randint(18, 25)
for b in range(n_balls):
deliveries_rows.append((
mid, bowler,
int(np.random.choice([0,1,2,4,6], p=[0.35,0.30,0.15,0.15,0.05])),
bool(np.random.random() < 0.06),
))
deliveries = spark.createDataFrame(
deliveries_rows,
["match_id", "bowler", "runs", "is_wicket"]
)
# Per-match bowler stats
bowler_match = (
deliveries
.groupBy("match_id", "bowler")
.agg(
F.sum("runs").alias("runs_conceded"),
F.count("*").alias("balls_bowled"),
F.sum(F.col("is_wicket").cast("int")).alias("wickets"),
)
.withColumn("economy",
F.round(F.col("runs_conceded") / (F.col("balls_bowled") / 6.0), 2))
)
# Window specs
bowler_time_window = (
Window.partitionBy("bowler")
.orderBy("match_id")
.rowsBetween(-2, 0) # 3-match rolling window
)
bowler_cumulative = (
Window.partitionBy("bowler")
.orderBy("match_id")
.rowsBetween(Window.unboundedPreceding, Window.currentRow)
)
# Rolling features
bowler_features = (
bowler_match
.withColumn("rolling_3m_economy",
F.round(F.avg("economy").over(bowler_time_window), 2))
.withColumn("rolling_3m_wicket_rate",
F.round(F.avg("wickets").over(bowler_time_window), 2))
.withColumn("season_wickets_to_date",
F.sum("wickets").over(bowler_cumulative))
)
# Validate: zero nulls in rolling columns (min_periods=1 from rowsBetween covers start)
for col_name in ["rolling_3m_economy","rolling_3m_wicket_rate","season_wickets_to_date"]:
null_count = bowler_features.filter(F.col(col_name).isNull()).count()
assert null_count == 0, f"{col_name} has {null_count} nulls"
print(f"Step 1 ✓: {bowler_features.count()} bowler-match records with rolling features")
bowler_features.orderBy("bowler","match_id").show(10, truncate=False)Step 2 — Broadcast Join, Pandas UDF Normalisation and Schema Check
Create a small venue conditions reference table (five rows), broadcast-join it onto the bowler features on `match_id`, and apply a Pandas UDF that min-max normalises the three rolling features across the full dataset in one vectorised pass. Validate the physical plan confirms `BroadcastHashJoin`, assert all normalised feature values are in [0, 1], and check the output schema against the expected `StructType` before proceeding to the Delta write.
Analogy🏏Cricket
🏏 Think of it like cricket: the broadcast join hands every analyst a copy of the five-line pitch report — dew, turn, average first-innings score — rather than mailing millions of bowler records to wherever the venue file lives; the physical-plan check confirms the field was actually set as signalled. The min-max normalisation is the scout's universal rating problem: to put five bowlers' economies on a common 0-to-1 scale you must first know the league's best and worst figures, which means seeing every bowler's numbers before scoring any single one. A scout rating one file at a time in isolation cannot do it — and that is exactly why the Pandas UDF processes the whole column in one vectorised pass rather than row by row. Asserting every value lands in [0, 1] is checking no rating fell off the scale, and the schema check is the team-sheet verification before submission. The payoff: features every downstream model can compare fairly, computed at batch speed instead of ball-by-ball.
python
# exercise_pyspark_features.py — Step 2: Join, normalise, validate
from pyspark.sql.functions import pandas_udf
import pandas as pd
# Small venue reference table — broadcast candidate
venue_ref = spark.createDataFrame([
(10001, "Wankhede", "flat", True),
(10002, "Chinnaswamy","flat", False),
(10003, "Eden Gardens","seaming",False),
(10004, "Chepauk", "turning",True),
(10005, "Wankhede", "flat", False),
], ["match_id","venue","pitch_type","dew_factor"])
# Broadcast join — venue_ref is tiny, eliminate shuffle
enriched_features = bowler_features.join(
F.broadcast(venue_ref), on="match_id", how="left"
)
# Verify BroadcastHashJoin in physical plan
plan_str = enriched_features._jdf.queryExecution().toString()
assert "BroadcastHashJoin" in plan_str or "BroadcastExchange" in plan_str, \
"Expected BroadcastHashJoin — check autoBroadcastJoinThreshold"
print("Step 2a ✓: BroadcastHashJoin confirmed in physical plan")
# Pandas UDF: min-max normalise rolling features across the full dataset
@pandas_udf(DoubleType())
def minmax_normalise(series: pd.Series) -> pd.Series:
mn, mx = series.min(), series.max()
if mx == mn:
return pd.Series([0.0] * len(series))
return (series - mn) / (mx - mn)
norm_features = (
enriched_features
.withColumn("economy_norm", minmax_normalise(F.col("rolling_3m_economy")))
.withColumn("wicket_rate_norm", minmax_normalise(F.col("rolling_3m_wicket_rate")))
.withColumn("season_wkts_norm", minmax_normalise(F.col("season_wickets_to_date").cast("double")))
)
# Assert all normalised values in [0, 1]
for col_name in ["economy_norm","wicket_rate_norm","season_wkts_norm"]:
out_of_range = norm_features.filter(
(F.col(col_name) < 0) | (F.col(col_name) > 1)
).count()
assert out_of_range == 0, f"{col_name} has {out_of_range} values outside [0,1]"
print("Step 2b ✓: all normalised features in [0, 1]")
# Schema validation
expected_cols = {"match_id","bowler","economy","wickets",
"rolling_3m_economy","economy_norm","venue","pitch_type"}
actual_cols = set(norm_features.columns)
assert expected_cols.issubset(actual_cols), f"Missing: {expected_cols - actual_cols}"
print(f"Step 2c ✓: schema validated — {len(actual_cols)} columns present")Step 3 — Delta Write with Constraints and Log Inspection
Write the normalised feature set to a Delta table, add CHECK constraints for normalised feature range and economy bounds, then attempt a write that violates both constraints to confirm enforcement. Inspect the Delta transaction log to verify the number of commits matches the number of write operations, and use time travel to confirm the table state before the constraint was added reflects the original data correctly.
Analogy🏏Cricket
🏏 Think of it like cricket: the CHECK constraint is the official scorebook's refusal to accept an impossible entry. A scorer simply cannot record a bowler conceding negative runs or an economy of 50 — the laws of the game bound what a legal entry can be, and the book rejects the nonsense at the moment of writing, not weeks later when the season table looks absurd. Deliberately submitting a violating write and watching it bounce is the umpire's pre-match test of the no-ball sensor: you prove the guard works before you rely on it. The transaction log is the match referee's report — one numbered entry per accepted operation, so counting commits against writes performed is reconciling the report against the day's play; any mismatch means something happened off the record. Time travel back to the pre-constraint state is re-reading the book exactly as it stood at tea. The payoff is a feature store like a certified scorecard: every number defensible, every change accounted for, nothing rewritten in secret.
python
# exercise_pyspark_features.py — Step 3: Delta write and constraints
from delta import DeltaTable
import os
FEATURE_PATH = "/tmp/ipl_features/bowler_features/"
# Select final feature columns for the Delta table
final_features = norm_features.select(
"match_id","bowler","venue","pitch_type","dew_factor",
"economy","wickets","balls_bowled",
"rolling_3m_economy","rolling_3m_wicket_rate","season_wickets_to_date",
"economy_norm","wicket_rate_norm","season_wkts_norm",
)
# Write version 0
final_features.write.format("delta").mode("overwrite").save(FEATURE_PATH)
print("Version 0: initial feature set written")
# Add CHECK constraints
spark.sql(f"""
ALTER TABLE delta.`{FEATURE_PATH}`
ADD CONSTRAINT valid_economy CHECK (economy >= 0 AND economy <= 36)
""")
spark.sql(f"""
ALTER TABLE delta.`{FEATURE_PATH}`
ADD CONSTRAINT valid_norm CHECK (
economy_norm >= 0.0 AND economy_norm <= 1.0
)
""")
print("Constraints added")
# Attempt to violate the economy constraint
bad_row = spark.createDataFrame(
[(10001, "BadBowler", "Wankhede", "flat", False,
-5.0, 0, 0, 0.0, 0.0, 0, 0.0, 0.0, 0.0)],
final_features.columns
)
try:
bad_row.write.format("delta").mode("append").save(FEATURE_PATH)
except Exception as e:
print(f"Constraint enforced ✓: {type(e).__name__}")
# Inspect transaction log
log_files = sorted(os.listdir(f"{FEATURE_PATH}_delta_log/"))
print(f"\nTransaction log entries: {len(log_files)} files")
for f in log_files:
print(f" {f}")
# Time travel: confirm version 0 (before constraints were added) is still readable
v0 = spark.read.format("delta").option("versionAsOf", 0).load(FEATURE_PATH)
print(f"\nVersion 0 row count: {v0.count()} (should match initial feature set)")
assert v0.count() == final_features.count(), "Time travel row count mismatch"
# Current version
current = spark.read.format("delta").load(FEATURE_PATH)
print(f"Current version row count: {current.count()}")
assert current.count() == v0.count(), "Constraint rejection left table unchanged ✓"
print("Step 3 ✓: Delta constraints enforced, time travel verified")