Capstone — Submit Big Data Streaming Project — Big Data & Distributed Computing

This is the final submission for Course 3: Big Data and Distributed Computing. You have built a complete streaming analytics platform for IPL stadium IoT events that integrates every module of the course: distributed computing concepts, HDFS and YARN fundamentals, Spark batch and streaming, PySpark joins and window functions, Delta Lake ACID transactions, Kafka producers and consumers with Avro schemas, Spark Structured Streaming with watermarks, and exactly-once idempotent sink writes. All integration test assertions must pass before submitting.

Before submitting, run the complete end-to-end integration test from the code block below. It executes all four pipeline stages sequentially and asserts every correctness invariant in one continuous verification flow. The integration test is the definitive acceptance criterion: a submission where all assertions pass represents a production-patterned, correctness-verified big data streaming pipeline that demonstrates mastery of every concept taught in this course.

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.

Final Integration Test

Run the complete integration test below. It re-executes the full pipeline end-to-end: re-generates IoT events, validates schema registration, produces to mock Kafka with partition routing, writes Parquet source files, runs the Structured Streaming windowed aggregations with `trigger(once=True)`, performs the anomaly detection MERGE, writes the raw landing Delta table, generates the session report, and asserts all invariants. Successful completion of all assertions confirms the pipeline is submission-ready.

Analogy🏏Cricket

🏏 Think of it like cricket: this test is the full dress-rehearsal match before a World Cup final — not the openers practising alone in one net and the bowlers in another, but a complete simulated match under match rules, because units that pass in isolation can still fail in combination. The four stages run in strict sequence for the same reason a match does: the ingestion is the innings that must be completed before the streaming aggregation — the scoring — can be finalised, which must precede the archive entry, which must precede the certified match report; you cannot write the report of an innings that never finished, and a failure in any stage invalidates everything downstream of it. Running with trigger(once=True) is playing one full simulated innings against a fixed, known set of deliveries, so every assertion has an exact expected answer. And the invariants — events conserved, replays changing nothing — are the umpires' books balancing at close of play. When the rehearsal match completes with every check green, the real submission holds no surprises.

python

# capstone_integration_test.py — Full end-to-end verification
# Run this file to verify all pipeline stages before submitting

print("=" * 60)
print("IPL IoT Streaming Capstone — Integration Test")
print("=" * 60)

# ── Stage 1: Kafka Producer ───────────────────────────────────────────────────
print("\n[Stage 1] Kafka Producer + Schema Registry")

# Assumes capstone_kafka_ingest.py has already been run
# Verify producer output using previously built mock_kafka object
assert kafka.total_messages("ipl-player-wearable") == 100, \
    f"Expected 100 wearable events"
assert kafka.total_messages("ipl-pitch-impact") == 100, \
    f"Expected 100 pitch events"
assert kafka.total_messages("ipl-crowd-sensor") == 100, \
    f"Expected 100 crowd events"
assert kafka.total_messages(IoTEventProducer.DLQ) == 1, \
    f"Expected 1 DLQ event (malformed player event)"
print(f"  Stage 1 ✓ — 300 events across 3 topics, 1 DLQ event")

# ── Stage 2: Spark Structured Streaming ──────────────────────────────────────
print("\n[Stage 2] Spark Structured Streaming + Delta")

from pyspark.sql import functions as F
from delta import DeltaTable

DELTA_BASE = "/tmp/iot_delta"

# Verify window tables exist and have rows
for name, path in [
    ("wearable_windows", f"{DELTA_BASE}/wearable_windows"),
    ("pitch_windows",    f"{DELTA_BASE}/pitch_windows"),
    ("crowd_windows",    f"{DELTA_BASE}/crowd_windows"),
]:
    count = spark.read.format("delta").load(path).count()
    assert count > 0, f"{name} Delta table is empty"
    print(f"  {name}: {count} window rows ✓")

# Verify fatigue alerts exist
alert_count = spark.read.format("delta").load(f"{DELTA_BASE}/wearable_windows") \
    .filter(F.col("is_fatigue_alert") == True).count()
assert alert_count > 0, "Expected at least one fatigue alert"
print(f"  Fatigue alerts detected: {alert_count} ✓")

# ── Stage 3: Delta Landing Zone ───────────────────────────────────────────────
print("\n[Stage 3] Delta Landing Zone")

LANDING = f"{DELTA_BASE}/raw_events"
landing_count = spark.read.format("delta").load(LANDING).count()
assert landing_count >= 300, f"Expected >= 300 raw events, got {landing_count}"
assert "processed" in spark.read.format("delta").load(LANDING).columns, \
    "Expected 'processed' column after schema evolution"
history_count = DeltaTable.forPath(spark, LANDING).history().count()
print(f"  Landing zone: {landing_count} rows, {history_count} Delta versions ✓")

# ── Stage 4: Session Report ───────────────────────────────────────────────────
print("\n[Stage 4] Batch Session Report")

wearable_windows = spark.read.format("delta").load(f"{DELTA_BASE}/wearable_windows")
pitch_windows    = spark.read.format("delta").load(f"{DELTA_BASE}/pitch_windows")

fatigue_summary = (
    wearable_windows
    .withColumn("fatigue_score",
        F.round((F.col("avg_hr") - 70) / 120, 3))
    .groupBy("match_id")
    .agg(F.round(F.avg("fatigue_score"),3).alias("avg_fatigue"),
         F.sum(F.col("is_fatigue_alert").cast("int")).alias("alerts"))
)
pitch_summary = (
    pitch_windows.groupBy("match_id")
    .agg(F.round(F.avg("avg_spin"),0).alias("avg_spin"))
)
session_report = (
    fatigue_summary
    .join(F.broadcast(pitch_summary), on="match_id", how="left")
    .orderBy("avg_fatigue", ascending=False)
)
assert session_report.count() == 3, \
    f"Expected 3 match rows in session report, got {session_report.count()}"

print("\nFinal Session Report:")
session_report.show(truncate=False)

# ── Summary ───────────────────────────────────────────────────────────────────
print("\n" + "=" * 60)
print("ALL INTEGRATION ASSERTIONS PASSED")
print("Course 3 Capstone Pipeline is Submission-Ready")
print("=" * 60)

Submission Checklist

Verify each item before submitting. Required items cause automatic deductions if absent; recommended items affect the architecture and code quality score. The checklist is ordered by pipeline stage — work through it sequentially to catch issues at the earliest possible stage, before they propagate forward to later assertions that depend on the output of earlier stages being correct.

Analogy🏏Cricket

🏏 Think of it like cricket: working through the checklist in pipeline order is the umpires' pre-match ritual, which is always done in strict sequence for a reason: pitch inspection before the toss, the toss before the team sheets, the team sheets before play — because a problem discovered late invalidates everything that followed it. A fault in your Stage 1 schema registration is like an ineligible player discovered in the 40th over: every ball bowled since is contaminated, and you replay the lot. Checking stage by stage catches the fault where it enters, the way a scorer reviewing an innings starts at over one — an error in the powerplay entry corrupts every cumulative total after it, so you fix it at the source rather than chasing its echoes through the death overs. The required-versus-recommended split mirrors the laws versus the spirit of the game: missing a required item forfeits marks automatically, while the recommended items are what separates a competent performance from a professional one. The payoff is walking into submission day the way a well-drilled side walks onto the field — nothing left to chance.

python

# Submission checklist — verify each before submitting

# [REQUIRED] Stage 1: Kafka Producer
# 1. All three Avro schemas registered with MockSchemaRegistry
# 2. Total events across all topics = 300 (100 per sensor type)
# 3. DLQ count = 1 (malformed player event injected in Step 2)
# 4. Partition routing verified: all events for same key in same partition

# [REQUIRED] Stage 2: Spark Structured Streaming
# 5. Three Delta tables created: wearable_windows, pitch_windows, crowd_windows
# 6. Each table has > 0 rows after trigger(once=True) run
# 7. wearable_windows contains 'is_fatigue_alert' column after MERGE
# 8. Fatigue alert count >= 1
# 9. Idempotency: MERGE run twice → alert count unchanged

# [REQUIRED] Stage 3: Delta Landing Zone
# 10. raw_events Delta table has >= 300 rows
# 11. raw_events has 'processed' column after schema evolution
# 12. Delta history shows >= 2 versions (initial write + schema evolution append)
# 13. CHECK constraint rejects metric_a < 0

# [REQUIRED] Stage 4: Session Report
# 14. session_report has exactly 3 rows (one per match)
# 15. session_report has columns: match_id, avg_fatigue, alerts, avg_spin
# 16. Full integration test passes with all assertions green

# [RECOMMENDED] Code Quality
# 17. Each streaming query has a unique checkpointLocation
# 18. All Delta MERGE operations use business keys, not batch IDs
# 19. watermark applied before groupBy in all streaming aggregations
# 20. F.broadcast used for all joins involving the small pitch_windows table

print("Checklist verification complete.")
print("Submit your project files to the SkillVeris capstone portal.")

What You Have Built: Over the six modules of Course 3, you have built and demonstrated mastery of the complete big data and distributed computing stack: CAP theorem, MapReduce, HDFS, and YARN; Spark RDDs, DataFrames, SQL, Parquet/ORC/Delta, partitioning and skew; PySpark joins, window functions, UDFs, MLlib, Spark UI tuning; Kafka topics, producers and consumers, consumer groups, Schema Registry, Kafka Connect; Spark Structured Streaming with watermarks and windowing; Apache Flink for event-at-a-time processing; and exactly-once semantics. You are equipped to design, build, tune, and operate production big data pipelines at any scale.