100% Free Forever
AI-Powered Learning
Industry Expert Content
Certificates & Badges
Learn At Your Own Pace
Big Data & Distributed Computing
55 minintermediate

Ingest Events via Kafka

This exercise implements Stage 1 of the capstone: a durable, schema-governed Kafka producer that publishes IoT sensor events from simulated IPL stadium devices. You will define Avro schemas for three sensor types — player wearables, pitch impact sensors, and crowd engagement sensors — register them with a mock Schema Registry, implement a partition-key routing strategy that groups events by sensor type and match ID, produce 300 events with correct routing, and verify the full producer pipeline with partition distribution and event count assertions.

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 exercise uses the `MockKafka` infrastructure from Module 4's practice exercise, extended with schema validation support that rejects events whose structure does not match the registered schema. This schema enforcement simulates the Confluent Schema Registry's role at the producer side, catching schema mismatches before they reach the Kafka topic and corrupt downstream consumers. All producer logic must handle schema validation errors by routing invalid events to a dead-letter queue rather than crashing.

Step 1 — Schema Definition and Mock Registry

Define three Avro-style schemas as Python dicts representing player wearable events, pitch impact events, and crowd engagement events. Implement a `MockSchemaRegistry` class that stores schemas by subject name, validates events against the registered schema, and enforces backward compatibility by rejecting schema changes that remove required fields. Register all three schemas and verify that a schema evolution adding an optional field is accepted while a breaking change removing a required field is rejected.

Analogy🏏Cricket
🏏 Think of it like cricket: defining the three schemas is writing the laws of scoring before the season starts: an official scorecard must carry these named columns, of these types, or it is not a scorecard. The registry is the MCC of the data platform — the body that holds the authoritative rulebook per record type and checks every proposed submission against it. The backward-compatibility rule is the crucial bit of lawmaking wisdom: a proposal to remove a required column — say, dropping 'extras' from the card — is rejected outright, because a century of analysis tools and historical comparisons read that column, and a format change that breaks every existing reader is not evolution, it is vandalism of the archive. Adding an optional column is fine, just as adding a new stat to the card leaves old cards readable. Verifying that the registry actually rejects the breaking change is testing the law before relying on it. The payoff: producers and consumers can evolve independently without ever breaking each other.
python
# capstone_kafka_ingest.py — Step 1: Schema definition and mock registry
import json
import hashlib
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from typing import Optional, Any
import random
import numpy as np

random.seed(42)
np.random.seed(42)

# ── Avro-style schema definitions ─────────────────────────────────────────────
PLAYER_WEARABLE_SCHEMA_V1 = {
    "type": "record",
    "name": "PlayerWearableEvent",
    "namespace": "com.ipl.iot",
    "fields": [
        {"name": "event_id",     "type": "long"},
        {"name": "match_id",     "type": "int"},
        {"name": "player_id",    "type": "int"},
        {"name": "player_name",  "type": "string"},
        {"name": "heart_rate",   "type": "int"},       # bpm
        {"name": "acceleration", "type": "float"},     # m/s²
        {"name": "running_dist", "type": "float"},     # metres since last event
        {"name": "ts_ms",        "type": "long"},      # Unix ms
    ]
}

PITCH_IMPACT_SCHEMA_V1 = {
    "type": "record",
    "name": "PitchImpactEvent",
    "namespace": "com.ipl.iot",
    "fields": [
        {"name": "event_id",     "type": "long"},
        {"name": "match_id",     "type": "int"},
        {"name": "delivery_id",  "type": "int"},
        {"name": "impact_x",     "type": "float"},     # pitch X coordinate (metres)
        {"name": "impact_y",     "type": "float"},     # pitch Y coordinate (metres)
        {"name": "ball_speed",   "type": "float"},     # km/h at bounce
        {"name": "spin_rate",    "type": "float"},     # rpm (0 if seam)
        {"name": "ts_ms",        "type": "long"},
    ]
}

CROWD_SENSOR_SCHEMA_V1 = {
    "type": "record",
    "name": "CrowdSensorEvent",
    "namespace": "com.ipl.iot",
    "fields": [
        {"name": "event_id",     "type": "long"},
        {"name": "match_id",     "type": "int"},
        {"name": "zone_id",      "type": "string"},    # stadium zone
        {"name": "decibels",     "type": "float"},
        {"name": "density",      "type": "float"},     # persons/m²
        {"name": "ts_ms",        "type": "long"},
    ]
}

# ── Mock Schema Registry ───────────────────────────────────────────────────────
class MockSchemaRegistry:
    """Simulates Confluent Schema Registry with backward-compatibility enforcement."""

    def __init__(self):
        self._subjects: dict[str, list[dict]] = {}   # subject → list of versions

    def register(self, subject: str, schema: dict) -> int:
        """Register a schema version. Returns version number (1-indexed)."""
        if subject not in self._subjects:
            self._subjects[subject] = []
        else:
            # Backward compatibility: new schema must not remove required fields
            prev = self._subjects[subject][-1]
            prev_fields = {f["name"] for f in prev["fields"]}
            new_fields  = {f["name"] for f in schema["fields"]}
            removed = prev_fields - new_fields
            if removed:
                raise ValueError(
                    f"Schema BACKWARD incompatible: removed required fields {removed}"
                )
        self._subjects[subject].append(schema)
        version = len(self._subjects[subject])
        print(f"  Registered schema '{subject}' v{version}")
        return version

    def validate(self, subject: str, event: dict) -> list[str]:
        """Validate event against the latest registered schema. Returns list of errors."""
        if subject not in self._subjects:
            return [f"Subject '{subject}' not registered"]
        schema  = self._subjects[subject][-1]
        errors  = []
        for fld in schema["fields"]:
            fname = fld["name"]
            if fname not in event:
                errors.append(f"Missing required field: '{fname}'")
            elif fld["type"] == "int" and not isinstance(event[fname], int):
                errors.append(f"Field '{fname}' must be int, got {type(event[fname]).__name__}")
            elif fld["type"] == "float" and not isinstance(event[fname], (int, float)):
                errors.append(f"Field '{fname}' must be float, got {type(event[fname]).__name__}")
        return errors

registry = MockSchemaRegistry()

# Register all three schemas
registry.register("player-wearable-value",  PLAYER_WEARABLE_SCHEMA_V1)
registry.register("pitch-impact-value",     PITCH_IMPACT_SCHEMA_V1)
registry.register("crowd-sensor-value",     CROWD_SENSOR_SCHEMA_V1)

# Verify: backward-compatible evolution accepted (add optional field)
PLAYER_WEARABLE_SCHEMA_V2 = {
    **PLAYER_WEARABLE_SCHEMA_V1,
    "fields": PLAYER_WEARABLE_SCHEMA_V1["fields"] + [
        {"name": "temperature", "type": ["null", "float"], "default": None}
    ]
}
try:
    v2 = registry.register("player-wearable-value", PLAYER_WEARABLE_SCHEMA_V2)
    print(f"  Evolution to v{v2} accepted ✓")
except ValueError as e:
    print(f"  Evolution rejected: {e}")

# Verify: breaking change rejected (remove required field)
BREAKING_SCHEMA = {"type":"record","name":"Test","namespace":"x",
    "fields": [{"name":"event_id","type":"long"}]}  # removed all other fields
try:
    registry.register("player-wearable-value", BREAKING_SCHEMA)
    print("  Breaking change accepted — ERROR")
except ValueError as e:
    print(f"  Breaking change correctly rejected ✓: {e}")

Step 2 — Producer with Routing, DLQ and Partition Verification

Implement the `IoTEventProducer` that generates 300 synthetic sensor events across the three types and three matches, routes each to the correct Kafka topic using match_id:sensor_type as the partition key, validates each event against its registered schema before producing, and routes invalid events to a DLQ topic. Verify the total event count, per-topic distribution, partition-routing correctness, and zero DLQ events for the clean data generation path before injecting a malformed event to confirm DLQ routing works correctly.

Analogy🏏Cricket
🏏 Think of it like cricket: the producer is the match-day data crew, and topic routing is the iron rule that each reading goes to its proper feed: heart-rate readings to the physio's channel, ball-impact data to the DRS channel, crowd noise to the operations channel — never one jumbled feed where the third umpire must wade through decibel readings to find ball-tracking. The match:sensor partition key is the filing rule that keeps one match's readings of one type together in arrival order, the way a single match's ball-tracking must stay in delivery sequence for a review to make sense. Validating each event against its registered schema before publishing is the boundary rope of data quality — checked at the source, not discovered downstream — and routing failures to the DLQ with the errors attached is putting the faulty reading in the review tray with the fault noted, so a glitching sensor never contaminates the official feed. The verification counts are the day's reconciliation: every reading accounted for, in the right feed, in order.
python
# capstone_kafka_ingest.py — Step 2: Producer with routing and DLQ

@dataclass
class MockMessage:
    topic:     str
    partition: int
    offset:    int
    key:       Optional[str]
    value:     dict

class MockKafka:
    N_PARTITIONS = 6

    def __init__(self):
        self.topics:    dict[str, dict[int, list[MockMessage]]] = defaultdict(
            lambda: defaultdict(list)
        )
        self.committed: dict[str, dict[str, dict[int, int]]] = defaultdict(
            lambda: defaultdict(lambda: defaultdict(int))
        )

    def produce(self, topic: str, key: Optional[str], value: dict) -> MockMessage:
        partition = self._partition_for_key(key)
        offset    = len(self.topics[topic][partition])
        msg = MockMessage(topic, partition, offset, key, value)
        self.topics[topic][partition].append(msg)
        return msg

    def _partition_for_key(self, key: Optional[str]) -> int:
        if key is None:
            return random.randint(0, self.N_PARTITIONS - 1)
        return int(hashlib.md5(key.encode()).hexdigest(), 16) % self.N_PARTITIONS

    def total_messages(self, topic: str) -> int:
        return sum(len(msgs) for msgs in self.topics[topic].values())

class IoTEventProducer:
    TOPICS = {
        "player": "ipl-player-wearable",
        "pitch":  "ipl-pitch-impact",
        "crowd":  "ipl-crowd-sensor",
    }
    DLQ = "ipl-iot-dlq"

    def __init__(self, kafka: MockKafka, registry: MockSchemaRegistry):
        self.kafka    = kafka
        self.registry = registry
        self.produced:  dict[str, int] = defaultdict(int)
        self.dlq_count: int = 0

    def _subject(self, sensor_type: str) -> str:
        return {"player":"player-wearable-value",
                "pitch": "pitch-impact-value",
                "crowd": "crowd-sensor-value"}[sensor_type]

    def publish(self, sensor_type: str, match_id: int, event: dict) -> None:
        subject = self._subject(sensor_type)
        errors  = self.registry.validate(subject, event)
        key     = f"{match_id}:{sensor_type}"
        if errors:
            self.kafka.produce(self.DLQ, key,
                               {"original": event, "errors": errors, "subject": subject})
            self.dlq_count += 1
        else:
            self.kafka.produce(self.TOPICS[sensor_type], key, event)
            self.produced[sensor_type] += 1

# Generate 300 synthetic IoT events
kafka    = MockKafka()
producer = IoTEventProducer(kafka, registry)

base_ts  = int(datetime(2024,4,20,19,30,0,tzinfo=timezone.utc).timestamp() * 1000)
MATCHES  = [10001, 10002, 10003]
PLAYERS  = [(1,"Rohit"),(2,"Kohli"),(3,"Bumrah"),(4,"Dhoni")]
ZONES    = ["North","South","East","West"]

event_id = 1
for i in range(100):   # 100 events per sensor type
    ts_ms   = base_ts + i * 1_000   # one event per second
    match   = random.choice(MATCHES)

    # Player wearable
    pid, pname = random.choice(PLAYERS)
    producer.publish("player", match, {
        "event_id":     event_id,    "match_id": match,
        "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),
        "ts_ms":        ts_ms,
    }); event_id += 1

    # Pitch impact
    producer.publish("pitch", match, {
        "event_id":    event_id, "match_id": match,
        "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, np.random.uniform(1000, 3000)])), 0),
        "ts_ms":       ts_ms,
    }); event_id += 1

    # Crowd sensor
    producer.publish("crowd", match, {
        "event_id": event_id, "match_id": match,
        "zone_id":  random.choice(ZONES),
        "decibels": round(float(np.random.normal(85, 12)), 1),
        "density":  round(float(np.random.uniform(0.5, 4.0)), 2),
        "ts_ms":    ts_ms,
    }); event_id += 1

print(f"\nProduced: {dict(producer.produced)}")
print(f"DLQ events: {producer.dlq_count}")
assert sum(producer.produced.values()) == 300
assert producer.dlq_count == 0
for stype, topic in IoTEventProducer.TOPICS.items():
    assert kafka.total_messages(topic) == 100, f"{topic}: expected 100"

# Inject malformed event — should route to DLQ
producer.publish("player", 10001, {"event_id": 9999, "heart_rate": "not_an_int"})
assert kafka.total_messages(IoTEventProducer.DLQ) == 1
print("DLQ routing for malformed event verified ✓")

# Partition routing: all events for same match_id:sensor_type → same partition
for match in MATCHES:
    for stype, topic in IoTEventProducer.TOPICS.items():
        key              = f"{match}:{stype}"
        expected_part    = kafka._partition_for_key(key)
        actual_partitions = {
            msg.partition
            for p_msgs in kafka.topics[topic].values()
            for msg in p_msgs
            if msg.key == key
        }
        if actual_partitions:
            assert actual_partitions == {expected_part}, \
                f"Key {key} routed to multiple partitions: {actual_partitions}"
print("Partition routing verified ✓")
print("\nStep 2 complete: 300 events produced across 3 sensor types, 3 topics")

Warning: The `MockSchemaRegistry.validate` function in this exercise checks field presence and primitive type correctness but does not validate union types, nested records, or enum constraints — a full Avro validator is required for production use. In production, the `confluent-kafka` library's `AvroSerializer` performs complete Avro validation using the `fastavro` library before serialising the message, and the Schema Registry enforces compatibility rules at registration time. Always use the `confluent-kafka` Avro serialiser rather than a hand-rolled validator for production pipelines.

Topic Naming Convention: Production Kafka deployments use a consistent topic naming convention that encodes the data domain, event type, and schema format. A common standard is `{domain}-{entity}-{eventtype}` in kebab-case: `ipl-player-wearable`, `ipl-pitch-impact`, `ipl-crowd-sensor`. The Schema Registry subject follows the pattern `{topic}-value` for value schemas and `{topic}-key` for key schemas. Consistent naming enables automated tooling — schema governance dashboards, lineage trackers, and data catalogues — to discover and link topics to their schemas without manual registration.

Lesson 32 of 35
0% complete