Streaming Systems Basics

• 1.5h: Concepts (events, windows, semantics, late data, joins).
• 2.5h: Worked examples below (run and tweak them).
• 1h: Build a mini pipeline using the mini project outline.
• 1h: Add schema evolution and dedup keys.
• 1h: Add basic monitoring metrics and alerts.
• 1h: Take the exam; revisit weak areas.

Goal: Send JSON events with event_time in UTC, consume, and print a parsed field.

# Producer (Python pseudocode)
import json, time, uuid, random, datetime as dt
# from kafka import KafkaProducer  # example library if available
producer = None  # assume producer created with idempotence enabled

users = ["u1","u2","u3"]
for i in range(5):
    evt = {
        "event_id": str(uuid.uuid4()),
        "user_id": random.choice(users),
        "event_time": dt.datetime.utcnow().isoformat() + "Z",
        "value": random.randint(1,100)
    }
    payload = json.dumps(evt).encode("utf-8")
    # producer.send("events", key=evt["user_id"].encode(), value=payload)
    print("Produced:", payload)
    time.sleep(0.2)

# Consumer (pseudocode)
# for msg in consumer.poll(...):
#     e = json.loads(msg.value)
#     print(e["user_id"], e["value"])  

Key points: include an event_id, use a partitioning key (user_id) for ordering, and encode event_time in UTC.

# Python-style Spark example (conceptual)
from pyspark.sql import functions as F

# events: key, event_time, value
# df = spark.readStream.format("kafka").option("subscribe","events").load()
# parsed = df.select(F.from_json(F.col("value").cast("string"), schema).alias("e")).select("e.*")

agg = (parsed
       .withColumn("ts", F.to_timestamp("event_time"))
       .groupBy(F.window("ts", "1 minute"), F.col("user_id"))
       .agg(F.sum("value").alias("sum_value")))

# query = agg.writeStream.outputMode("update").format("console").start()

Key points: Tumbling windows are fixed-size, non-overlapping. Use event time for accurate aggregation.

# Allow events up to 10 minutes late
with_wm = (parsed
    .withColumn("ts", F.to_timestamp("event_time"))
    .withWatermark("ts", "10 minutes")
    .groupBy(F.window("ts", "5 minutes"), "user_id")
    .agg(F.count("*").alias("cnt")))

Watermarks let the engine drop state for windows older than watermark, bounding memory while tolerating definable lateness.

# Join click events to user profile updates within 15 minutes
clicks = parsed.withColumn("ts", F.to_timestamp("event_time")).withWatermark("ts","15 minutes")
profiles = profiles_stream.withColumn("ts2", F.to_timestamp("event_time")).withWatermark("ts2","15 minutes")

joined = (clicks.join(profiles, F.expr("clicks.user_id = profiles.user_id AND clicks.ts BETWEEN ts2 - interval 15 minutes AND ts2 + interval 15 minutes"), "inner"))

Require watermarks on both sides and a bounded time condition to keep state finite.

# Pattern: deduplicate by event_id and upsert
# 1) Ingest with at-least-once delivery.
# 2) In sink, MERGE by event_id (or (user_id, version)).
# 3) Make writes transactional if supported (e.g., a table with ACID semantics).

# Pseudocode transformation step
clean = parsed.dropDuplicates(["event_id"])  # engine-managed dedup if available
# sink.writeStream.option("mergeKey","event_id").format("delta").start()

Exactly-once is often achieved end-to-end by combining idempotent producers, transactional consumption, and idempotent upserts in sinks.

# Lag = latest_offset - committed_offset, aggregated across partitions
# Example numbers per partition:
# p0: latest 1200, committed 1100 -> lag 100
# p1: latest 1500, committed 1490 -> lag 10
# Total lag: 110

# Heuristic: if (lag / events_per_min) > target_latency_minutes, scale consumers

Track both consumer lag and end-to-end latency (event_time to output_time). Alert when thresholds are breached.

• Print sample records with event_time and processing_time to confirm ordering assumptions.
• Turn on aggregation state metrics; watch watermark progression.
• Force a duplicate batch and confirm idempotent behavior in the sink.
• Stress test partition skew: randomize keys and check throughput per partition.
• Simulate schema change: add an optional field and verify consumers continue to work.

• Aggregations correct despite 10% late events up to 10 minutes.
• Duplicate events do not change final results (idempotent).
• Join does not grow unbounded state; watermarks advance.
• Lag stays below a set threshold under moderate load.

• Add a 30-second sliding window metric alongside the tumbling window.
• Derive an alert stream when value spikes above a dynamic threshold (e.g., rolling 95th percentile).
• Expose basic health metrics: input rate, processed rate, lag, watermark age.