Stream Processing

Chapter 1: Event Streams

Before we can process a stream, we need to define what a stream is. Forget queues and brokers for a moment. The concept is simpler than the infrastructure.

An event is an immutable record of something that happened at a point in time. "User 42 clicked the buy button at 14:03:27.391 UTC." "Sensor 7 reported temperature 23.4C at 14:03:27.500 UTC." "Order 9001 was shipped at 14:03:28.012 UTC." Each event has a timestamp, a type, and a payload. Once it happened, it cannot un-happen. You can record a compensating event (a refund reverses a charge), but the original event remains in the record forever.

A stream (also called an event stream, event log, or event feed) is an unbounded, append-only, ordered sequence of events. "Unbounded" is the key word: unlike a file that has a beginning and an end, a stream has a beginning but no end. New events keep arriving indefinitely.

Producers, Streams, and Consumers

The basic architecture has three roles:

This decoupling is powerful. The producer does not slow down if a consumer is behind. A new consumer can join at any time and read from the beginning. A slow consumer does not block a fast one. The stream is the buffer, the ordering guarantee, and the persistent record all in one.

Multiple Consumers, One Stream

The simulation below shows a stream with three consumers reading at different speeds. Consumer A is fast (real-time analytics). Consumer B is medium (search indexing). Consumer C is slow (batch export to a data warehouse). Each tracks its own position in the stream.

Event Sourcing Preview

Notice something profound: if we store every event, we can reconstruct any state at any point in time. The current balance of account 42 is just the sum of all deposits minus all withdrawals up to now. The balance at 3 PM yesterday is the same sum, but stopping at that timestamp. This is event sourcing, and we will explore it in depth in Chapter 5.

python
# An event: immutable, timestamped, typed
from dataclasses import dataclass
from datetime import datetime

@dataclass(frozen=True)  # frozen = immutable
class Event:
    timestamp: datetime
    event_type: str      # "deposit", "withdrawal", "transfer"
    payload: dict        # {"account": 42, "amount": 100.00}

# A stream: append-only, ordered
class EventStream:
    def __init__(self):
        self.log = []         # append-only list of events
        self.consumers = {}   # consumer_id -> offset

    def append(self, event: Event) -> int:
        offset = len(self.log)
        self.log.append(event)
        return offset         # sequential position

    def read(self, consumer_id: str, batch_size: int = 1):
        offset = self.consumers.get(consumer_id, 0)
        events = self.log[offset:offset + batch_size]
        self.consumers[consumer_id] = offset + len(events)
        return events

    def seek(self, consumer_id: str, offset: int):
        """Reset a consumer's position — enables replay."""
        self.consumers[consumer_id] = offset

Chapter 2: Message Brokers

Events need to get from producers to consumers. In theory, you could have producers write directly to consumers (a webhook, a TCP socket, a UDP multicast). In practice, this creates tight coupling: the producer must know every consumer's address, handle retries if a consumer is down, and buffer events if a consumer is slow. This does not scale.

A message broker (also called a message queue or messaging middleware) sits between producers and consumers. Producers send events to the broker. Consumers pull events from the broker. The broker handles durability, buffering, delivery guarantees, and fan-out.

There are two fundamentally different models for message brokers, and understanding the difference is critical for system design interviews.

Model 1: Traditional Message Queues

Systems like RabbitMQ, ActiveMQ, and Amazon SQS follow this model. The broker maintains a queue of messages. When a consumer acknowledges a message, the broker deletes it. Key properties:

Model 2: Log-Based Brokers

Systems like Apache Kafka, Amazon Kinesis, and Apache Pulsar follow this model. The broker is an append-only log. Messages are never deleted (until a configurable retention period expires). Key properties:

Side-by-Side Comparison

The simulation below shows both models handling the same events. On the left, a traditional message queue delivers each message to one of two consumers and deletes it. On the right, a log-based broker appends each message and lets consumers track their own offset. Watch what happens when a consumer crashes and recovers.

Notice what happens after recovery. In the traditional queue, the crashed consumer lost the unacknowledged message — it may be redelivered (at-least-once) or lost (at-most-once), depending on configuration. In the log-based broker, the consumer simply resumes from its last committed offset. No message is ever lost because the log persists regardless of consumer state.

When to Use Which

text
Use a traditional message queue (RabbitMQ, SQS) when:
- Task distribution: "process this image", "send this email"
- You need per-message routing (complex topic/exchange patterns)
- Message order does not matter
- You do not need replay
- Consumer count >> partition count (thousands of workers)

Use a log-based broker (Kafka, Kinesis) when:
- Event sourcing: "this thing happened"
- Multiple independent consumers need the same data
- You need replay (reprocess after a bug fix)
- Ordering within a key matters (all events for user X in order)
- You are building CDC, analytics, or materialized views

Chapter 3: Apache Kafka

Kafka is the log-based broker that won. Originally built at LinkedIn in 2011 to handle their activity stream (page views, clicks, searches — 1.4 trillion messages per day at peak), it is now the backbone of event-driven architectures at Netflix, Uber, Airbnb, and thousands of other companies. Let us understand exactly how it works.

Topics and Partitions

A topic is a named stream of events. "orders", "user-clicks", "sensor-readings" — each is a topic. But a single topic on a single machine would be a bottleneck. So Kafka splits each topic into partitions.

A partition is an ordered, append-only log stored on a single broker. Each event in a partition gets a monotonically increasing offset (0, 1, 2, 3...). Ordering is guaranteed within a partition but NOT across partitions.

When a producer sends an event, it must go to exactly one partition. Two strategies:

Consumer Groups

A consumer group is a set of consumers that cooperate to consume a topic. Kafka assigns each partition to exactly one consumer in the group. If you have 6 partitions and 3 consumers in a group, each consumer gets 2 partitions. If a consumer crashes, its partitions are reassigned to the remaining consumers — this is called a rebalance.

Multiple consumer groups can independently consume the same topic. Consumer group "fraud-detection" reads the "transactions" topic. Consumer group "analytics" reads the same topic. They do not interfere with each other. Each group tracks its own offsets.

Interactive Kafka Cluster

The simulation below shows a Kafka cluster with 3 brokers and 1 topic split into 6 partitions. Send messages with keys and watch them hash to partitions. Add consumers to a group and watch partition assignment. Remove a consumer and watch rebalancing.

Offsets and Commit

Each consumer tracks its offset — the position of the last event it has processed in each partition. Periodically (or after each event), the consumer commits its offset back to Kafka (stored in an internal topic called __consumer_offsets).

If a consumer crashes and restarts, it resumes from the last committed offset. This creates a delivery guarantee trade-off:

Kafka Producer and Consumer in Python

python
from kafka import KafkaProducer, KafkaConsumer
import json

# --- Producer ---
producer = KafkaProducer(
    bootstrap_servers=['broker1:9092', 'broker2:9092'],
    key_serializer=lambda k: k.encode('utf-8'),
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',           # wait for all replicas to acknowledge
    retries=3,             # retry on transient failures
    linger_ms=5,           # batch messages for 5ms for throughput
)

# Key determines partition: all user_42 events go to the same partition
producer.send(
    topic='transactions',
    key='user_42',
    value={'amount': 99.99, 'merchant': 'coffee_shop'}
)
producer.flush()  # block until all messages are sent

# --- Consumer ---
consumer = KafkaConsumer(
    'transactions',
    bootstrap_servers=['broker1:9092'],
    group_id='fraud-detection',   # consumer group name
    auto_offset_reset='earliest',  # start from beginning if no offset
    enable_auto_commit=False,     # manual commit for at-least-once
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
)

for msg in consumer:
    # msg.topic, msg.partition, msg.offset, msg.key, msg.value
    result = run_fraud_model(msg.value)
    if result.is_fraud:
        alert(msg.key, msg.value)
    consumer.commit()  # commit after successful processing

Chapter 4: Change Data Capture

Your application writes to PostgreSQL. Your search team wants every product update in Elasticsearch for full-text search. Your analytics team wants every order in the data warehouse. Your caching layer wants every user profile change in Redis. How do you keep all these systems in sync?

The naive approach: after writing to PostgreSQL, also write to Elasticsearch, the warehouse, and Redis. This is called dual writes, and it is a trap. If the PostgreSQL write succeeds but the Elasticsearch write fails, your systems are inconsistent. Even if you retry, the order of writes across systems is not guaranteed — one system might see update A then B, while another sees B then A. Dual writes are not atomic and do not preserve ordering.

Change Data Capture (CDC) is the solution. Instead of having the application write to multiple systems, you make the database the single source of truth and capture every change from its internal log. PostgreSQL has the Write-Ahead Log (WAL). MySQL has the binlog. These logs already record every INSERT, UPDATE, and DELETE — the database uses them for crash recovery and replication. CDC reads these logs and publishes the changes as events to a stream (usually Kafka).

How CDC Works

Log Compaction

Kafka topics have a retention period (default 7 days). After 7 days, old events are deleted. But what if a new consumer joins and needs the current state, not just the last 7 days of changes?

Log compaction solves this. Instead of deleting events by age, Kafka keeps only the latest event for each key. If user_42 has been updated 500 times, the compacted log retains only the most recent update. A new consumer can read the compacted log to build a complete snapshot of current state, then switch to the uncompacted tail for real-time updates.

To delete a record, the producer sends a tombstone — a message with the key but a null value. During compaction, the tombstone removes the key from the log.

Animated CDC Pipeline

Debezium Configuration

json
{
  "name": "orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres-primary",
    "database.port": "5432",
    "database.user": "debezium",
    "database.dbname": "ecommerce",
    "database.server.name": "prod-db",
    "table.include.list": "public.orders,public.products",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "dbz_publication",
    "topic.prefix": "prod-db",
    "snapshot.mode": "initial"
  }
}

Chapter 5: Event Sourcing

CDC captures changes that the application made to a database. The application writes "current state" (UPDATE accounts SET balance = 150 WHERE id = 42), and CDC derives the event stream from those mutations. Event sourcing flips this: the application writes events directly ("user 42 deposited $50"), and current state is derived from the event log.

Think of it like accounting. A traditional database is like a balance sheet: it tells you "Account 42 has $150 right now." Event sourcing is like a general ledger: it records every transaction. "Account 42 opened with $0. Deposited $100. Withdrew $20. Deposited $70." The balance sheet is derived from the ledger, not the other way around.

Why Store Events Instead of State?

CQRS: Command Query Responsibility Segregation

Event sourcing naturally leads to CQRS: separate the write side (accepting commands, producing events) from the read side (querying derived views).

Interactive Event Log

The simulation below is an event-sourced bank account. Append events (deposits, withdrawals) to the log on the left, and watch the derived balance update on the right. Hit "Replay from Scratch" to rebuild state from the event log.

python
from dataclasses import dataclass, field
from typing import List
import time

@dataclass(frozen=True)
class AccountEvent:
    event_type: str    # "created", "deposited", "withdrawn"
    amount: float
    timestamp: float = field(default_factory=time.time)

class EventSourcedAccount:
    def __init__(self, account_id: str):
        self.account_id = account_id
        self.events: List[AccountEvent] = []
        self._balance = 0.0  # derived state (cache)

    def deposit(self, amount: float):
        assert amount > 0
        event = AccountEvent("deposited", amount)
        self.events.append(event)
        self._balance += amount  # update cache

    def withdraw(self, amount: float):
        assert 0 < amount <= self._balance
        event = AccountEvent("withdrawn", amount)
        self.events.append(event)
        self._balance -= amount

    def replay(self) -> float:
        """Rebuild balance from events. The source of truth."""
        balance = 0.0
        for e in self.events:
            if e.event_type == "deposited":
                balance += e.amount
            elif e.event_type == "withdrawn":
                balance -= e.amount
        self._balance = balance
        return balance

    def balance_at(self, timestamp: float) -> float:
        """Temporal query: what was the balance at a point in time?"""
        balance = 0.0
        for e in self.events:
            if e.timestamp > timestamp:
                break
            if e.event_type == "deposited":
                balance += e.amount
            elif e.event_type == "withdrawn":
                balance -= e.amount
        return balance

Chapter 6: Stream Processing Operations

In batch processing, aggregation is straightforward: "count all the clicks in yesterday's log." The dataset is bounded — you know when it starts and when it ends. In stream processing, the dataset is unbounded. Events keep arriving forever. You cannot "count all the clicks" because there is no "all." You need a way to slice the infinite stream into finite chunks for aggregation. This is called windowing.

Four Types of Windows

A window is a time interval over which you compute an aggregate (count, sum, average, max). There are four types, and they differ in how they slice time.

Watermarks

A watermark is the system's estimate of "all events with timestamps up to time T have probably arrived." When the watermark advances past the end of a window, the system fires the window's computation and emits results.

Watermarks are a heuristic. They can be:

Late events arrive after the watermark has passed their window. Strategies: (1) Drop them. (2) Refire the window computation with updated results. (3) Route them to a side output for manual handling. Most production systems use a combination: allow late events up to an allowed lateness threshold, and drop anything beyond that.

Interactive Windowing Demo

This is the showcase simulation. Events arrive on a timeline. Choose a window type and size. Watch events get grouped into windows and aggregations computed per window. Late events (after the watermark) are highlighted.

Windowing in Apache Flink

python
# Apache Flink Python API (PyFlink) — tumbling window aggregation
from pyflink.table import EnvironmentSettings, TableEnvironment
from pyflink.table.window import Tumble
from pyflink.table.expressions import col, lit

env_settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(env_settings)

# Read from Kafka topic
t_env.execute_sql("""
    CREATE TABLE clicks (
        user_id STRING,
        url STRING,
        click_time TIMESTAMP(3),
        WATERMARK FOR click_time AS click_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user-clicks',
        'properties.bootstrap.servers' = 'broker:9092',
        'format' = 'json'
    )
""")

# Tumbling window: count clicks per user per 5-minute window
result = t_env.from_path('clicks') \
    .window(Tumble.over(lit(5).minutes).on(col('click_time')).alias('w')) \
    .group_by(col('user_id'), col('w')) \
    .select(
        col('user_id'),
        col('w').start.alias('window_start'),
        col('w').end.alias('window_end'),
        col('url').count.alias('click_count')
    )

Chapter 7: Stream Joins

In batch processing, joining two datasets is straightforward: both are bounded, both are fully available, and you can sort-merge or hash-join them. In stream processing, data arrives continuously and you cannot "wait for all the data" because it never ends. Stream joins are fundamentally harder, and there are three types — each with different semantics and implementation costs.

Type 1: Stream-Stream Join

You have two event streams and you want to match events from one with events from the other, within a time window. The canonical example: an ad platform wants to match ad impressions (user saw an ad) with ad clicks (user clicked the ad), where the click must happen within 1 hour of the impression.

Implementation: maintain a buffer for each stream. When an impression arrives, store it in the impression buffer. When a click arrives, search the impression buffer for a matching impression within the time window. Emit a joined result. Expire entries from the buffer when they fall outside the window.

Type 2: Stream-Table Join (Enrichment)

You have an event stream and you want to enrich each event with data from a table. Example: a "user activity" stream where each event has a user_id, and you want to add the user's name, country, and subscription tier from the user profile table.

Two approaches:

In practice, the local table copy (via CDC) is almost always the right choice for high-throughput systems. The table is typically much smaller than the stream and fits in memory.

Type 3: Table-Table Join (Materialized View)

Both inputs are CDC streams (representing tables), and the output is a continuously updated join. Example: a "orders" table and a "products" table. You want a materialized view that joins them: for each order, include the product name and price.

When an order is inserted, look up the product. When a product's price changes, update all orders that reference it. This is a live, continuously maintained join — the same thing a database does internally for a materialized view, but implemented at the application level using streams.

Three Join Types Side-by-Side

python
# Stream-stream join: match impressions with clicks within 1 hour
from collections import defaultdict
import time

class StreamStreamJoin:
    def __init__(self, window_seconds: int):
        self.window = window_seconds
        self.left_buffer = defaultdict(list)   # key -> [(timestamp, event)]
        self.right_buffer = defaultdict(list)

    def add_left(self, key, ts, event):
        self._evict(self.left_buffer, key, ts)
        self.left_buffer[key].append((ts, event))
        # Check right buffer for matches
        matches = []
        for rts, revt in self.right_buffer.get(key, []):
            if abs(ts - rts) <= self.window:
                matches.append((event, revt))
        return matches

    def add_right(self, key, ts, event):
        self._evict(self.right_buffer, key, ts)
        self.right_buffer[key].append((ts, event))
        matches = []
        for lts, levt in self.left_buffer.get(key, []):
            if abs(ts - lts) <= self.window:
                matches.append((levt, event))
        return matches

    def _evict(self, buffer, key, current_ts):
        """Remove entries outside the join window."""
        if key in buffer:
            buffer[key] = [(t, e) for t, e in buffer[key]
                           if current_ts - t <= self.window]

# Usage: match ad impressions with clicks within 1 hour
join = StreamStreamJoin(window_seconds=3600)
# Impression arrives
matches = join.add_left("ad_123", time.time(), {"type": "impression"})
# Later, click arrives
matches = join.add_right("ad_123", time.time(), {"type": "click"})
# matches = [(impression_event, click_event)]

Chapter 8: Fault Tolerance

A batch job that fails halfway through is easy to recover: delete the partial output and re-run the whole job. The input is immutable, the output is deterministic, and the job is bounded. Stream processing is harder. The job runs forever. It accumulates state (window aggregates, join buffers, counters). If it crashes, you cannot just "re-run from the beginning" — that would take hours or days. And you must guarantee that each event is processed correctly despite failures. Not once. Not zero times. Exactly once.

The Three Delivery Guarantees

Strategy 1: Idempotent Writes

Design your output operations so that re-processing an event produces the same result as processing it once.

To make non-idempotent operations safe, include a deduplication key (event ID or offset) in the output. Before writing, check if that key already exists. If it does, skip the write.

Strategy 2: Kafka Transactions

Kafka provides transactional producers that atomically write to multiple partitions and commit consumer offsets in a single transaction. Either all writes succeed and the offset is committed, or none of them do.

python
# Kafka transactional producer — exactly-once semantics
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers=['broker:9092'],
    transactional_id='fraud-processor-1',  # unique per instance
    enable_idempotence=True,               # required for transactions
)
producer.init_transactions()

try:
    producer.begin_transaction()
    # Process input events and produce output
    for msg in batch:
        result = process(msg.value)
        producer.send('fraud-alerts', key=msg.key, value=result)
    # Commit input offsets as part of the transaction
    producer.send_offsets_to_transaction(
        offsets={tp: offset for tp, offset in current_offsets.items()},
        group_id='fraud-detection'
    )
    producer.commit_transaction()
except Exception:
    producer.abort_transaction()  # rollback everything

Strategy 3: Chandy-Lamport Checkpointing (Flink)

Apache Flink uses a distributed snapshotting algorithm based on the Chandy-Lamport protocol. The idea:

Flink Checkpointing in Action

The simulation below shows a Flink pipeline with 3 operators processing events. A checkpoint barrier flows through the pipeline, each operator saves its state. Then a failure occurs — watch the system restore from the checkpoint and replay.

Design Challenge: Fraud Detection with Exactly-Once

Design a real-time fraud detection system that processes credit card transactions from Kafka, applies a fraud model, and writes alerts to a "fraud-alerts" topic with exactly-once semantics. Consider: what happens if the fraud model has side effects (e.g., incrementing a "suspicious activity count" per user)?

text
Design sketch:

1. Input: Kafka topic "transactions" (partitioned by card_id)
2. Flink job with:
   - Source: KafkaSource with exactly-once checkpoint mode
   - Operator 1: Enrich with user profile (stream-table join via CDC)
   - Operator 2: Windowed aggregation (count transactions per card in 5-min window)
   - Operator 3: Fraud model (flag if count > threshold or model score > 0.9)
   - Sink: KafkaTransactionalSink to "fraud-alerts" topic

3. Checkpointing: every 5 seconds to S3
4. The "suspicious activity count" per user MUST be part of Flink's
   managed state (not an external database), so it is included in
   checkpoints and restored correctly on failure.
5. If count is in an external DB, use idempotent writes:
   SET suspicious_count[user] = computed_value (not INCREMENT)

Chapter 9: Interview Arsenal

This chapter compresses every stream processing concept into interview-ready form. Organized by the five interview dimensions: concept, design, code, debug, and frontier.

Concept Cheat Sheet

System Design Questions

Debug Scenarios

Coding Drill: Tumbling Window Aggregator

python
from collections import defaultdict
import time

class TumblingWindowAggregator:
    """Count events per key per tumbling window."""

    def __init__(self, window_size_seconds: int):
        self.window_size = window_size_seconds
        # (window_start, key) -> count
        self.windows = defaultdict(int)
        self.emitted = set()  # track which windows have been emitted

    def _window_start(self, event_time: float) -> float:
        """Assign event to its tumbling window."""
        return event_time - (event_time % self.window_size)

    def process(self, key: str, event_time: float):
        ws = self._window_start(event_time)
        self.windows[(ws, key)] += 1

    def fire_windows(self, watermark: float):
        """Emit results for all windows whose end <= watermark."""
        results = []
        for (ws, key), count in list(self.windows.items()):
            window_end = ws + self.window_size
            if window_end <= watermark and (ws, key) not in self.emitted:
                results.append({
                    'window_start': ws,
                    'window_end': window_end,
                    'key': key,
                    'count': count,
                })
                self.emitted.add((ws, key))
        return results

# Example usage
agg = TumblingWindowAggregator(window_size_seconds=60)
agg.process("url_a", 1000.0)  # window [960, 1020)
agg.process("url_a", 1010.0)  # same window
agg.process("url_b", 1005.0)  # same window, different key
agg.process("url_a", 1025.0)  # next window [1020, 1080)

# Watermark advances past 1020 -> fire the first window
results = agg.fire_windows(watermark=1025.0)
# [{'window_start': 960, 'window_end': 1020, 'key': 'url_a', 'count': 2},
#  {'window_start': 960, 'window_end': 1020, 'key': 'url_b', 'count': 1}]

Coding Drill: Idempotent Event Processor

python
import hashlib, json

class IdempotentProcessor:
    """Process events exactly once using a deduplication log."""

    def __init__(self):
        self.seen = set()       # set of processed event IDs
        self.state = {}          # application state (e.g., user balances)

    def _event_id(self, event: dict) -> str:
        """Deterministic ID from event content."""
        raw = json.dumps(event, sort_keys=True).encode()
        return hashlib.sha256(raw).hexdigest()

    def process(self, event: dict) -> bool:
        eid = self._event_id(event)
        if eid in self.seen:
            return False  # already processed, skip

        # Process the event
        user = event['user_id']
        amount = event['amount']
        self.state[user] = self.state.get(user, 0) + amount

        self.seen.add(eid)
        return True  # successfully processed

# Same event processed twice -> only applied once
proc = IdempotentProcessor()
event = {'user_id': '42', 'amount': 100, 'ts': 1234567890}
proc.process(event)  # True  — applied, balance = 100
proc.process(event)  # False — skipped, balance still 100

Chapter 10: Connections

Stream processing does not exist in isolation. It connects to nearly every topic in distributed systems. Here is how this chapter relates to the rest of DDIA and beyond.

Where Stream Processing Fits

Comparison: Stream Processing Frameworks

Key Papers and Reading