Streaming Fundamentals

Batch vs streaming

There are two fundamental shapes for moving data through a pipeline, and almost everything else is a detail on top of one of them.

• Batch — process data on a schedule. Every hour (or night, or 15 minutes) a job runs, reads everything that piled up since last time, transforms it, writes results, and stops. This is what your dbt models and ingestion jobs do.
• Streaming — process data as it arrives. There's no "run"; a long-lived program sits there reacting to each event within milliseconds of it happening.

The trade is latency versus simplicity. Batch is dramatically simpler: a job either ran or it didn't, you can re-run it, you can inspect its input and output as files or tables. Streaming buys you freshness — seconds instead of hours — but you pay with always-on infrastructure, harder debugging, and a whole vocabulary of new failure modes.

The log abstraction

Strip away the branding and Kafka, Redpanda, Kinesis, and Pulsar are all the same idea: a log. Not "log" as in application logging — log as in an append-only, ordered sequence of events. New events are only ever added to the end; existing events are never modified. Each event gets a position number as it lands, and those numbers only go up.

Two terms carry all the weight here:

• A topic is a named stream — gpu-usage, payments, clicks. It's the logical channel producers write to and consumers read from.
• A partition is a slice of a topic's log. A topic is split into N partitions so the work can be spread across machines and readers. The catch: ordering is only guaranteed within a single partition. Events in partition 0 are strictly ordered; an event in partition 0 and an event in partition 1 have no defined order relative to each other.

That last point drives a key design choice: if you need all events for one GPU to stay in order, you give them the same key (the GPU id), and the log routes same-key events to the same partition. Different GPUs can spread across partitions for parallelism while each individual GPU's history stays ordered.

Producers, consumers & offsets

Two roles sit on either side of the log.

• A producer writes events to a topic. It's fire-and-(mostly)-forget: append to the end, move on.
• A consumer reads events from a topic, in order, one partition at a time.

The single most important concept on the consumer side is the offset. An offset is just a consumer's bookmark — the position number of the next event it intends to read. The log doesn't track who has read what; each consumer remembers its own offset. This is the whole trick, and it has two huge consequences:

• Replay. Because the events are still on disk and your position is just a number, you can move your bookmark backward — set the offset to 0 and re-read the entire history. Reprocessing a bug-fixed transformation over last week's events is as simple as rewinding. (You felt the value of this idempotent re-run mindset in the lakehouse; the log makes it native.)
• At-least-once delivery. A consumer reads an event, processes it, and then commits its new offset. If it crashes after processing but before committing, on restart it re-reads that event — so you might see an event more than once, but never lose one. This is the at-least-once guarantee, and it's why downstream logic must be idempotent — exactly the delivery-semantics idea from Foundations ch.11.

When one consumer can't keep up, you run several in a consumer group. The log hands each partition to exactly one member of the group, so three consumers can split a three-partition topic and process in parallel — while the group as a whole still reads every event exactly once between them.

Event-time & windows

When you aggregate a stream — "GPU-hours used per minute" — you have to bucket events into time windows. The subtle, important question is: which clock?

• Event-time — when the event actually happened (the timestamp the producer stamped on it).
• Processing-time — when your consumer happened to read it.

These differ because the network is messy. An event generated at 12:00:59 might not reach your consumer until 12:01:03 — a phone was offline, a queue backed up, a retry fired. If you bucket by processing-time, that event lands in the 12:01 window even though it belongs to 12:00. Your "usage at 12:00" number is now wrong, and worse, it's wrong in a way that changes depending on how busy your system was. You almost always want to window by event-time, so the same events always produce the same answer no matter when you process them.

Which raises the hard part: late data. If an event for the 12:00 window can show up at 12:01:03, when is the 12:00 window safe to close and emit? Close too early and you drop late events; wait forever and you never emit a result. Streaming engines solve this with a watermark — a moving marker that asserts "I believe I've now seen all events up to time T." When the watermark passes the end of a window, the engine finalizes it; events arriving after the watermark are flagged as late and handled by an explicit policy (drop, or re-open and correct). You met windows conceptually in Foundations; here you can see exactly why event-time is the honest clock to bucket by.

Hands-on: a live stream end to end

Time to make it real. We'll bring up Redpanda, write a producer that emits GPU-usage events to a topic, and a consumer that reads and counts them live.

1. Bring up Redpanda. Save this as docker-compose.streaming.yml in your repo:

services:
  redpanda:
    image: redpandadata/redpanda:v24.2.7
    container_name: redpanda
    command:
      - redpanda start
      - --smp 1                       # 1 CPU core — plenty for local
      - --overprovisioned
      - --kafka-addr PLAINTEXT://0.0.0.0:9092
      - --advertise-kafka-addr PLAINTEXT://localhost:9092
    ports:
      - "9092:9092"                   # the Kafka API your Python client talks to

docker compose -f docker-compose.streaming.yml up -d
docker compose -f docker-compose.streaming.yml ps   # redpanda should be "Up"

2. The producer. Save as producer.py. It emits one GPU-usage event per second, keyed by GPU id so each GPU's events stay ordered:

import json, time, random
from datetime import datetime, timezone
from confluent_kafka import Producer

producer = Producer({"bootstrap.servers": "localhost:9092"})
TOPIC = "gpu-usage"
GPUS = ["gpu-a100-01", "gpu-a100-02", "gpu-h100-01"]

def delivered(err, msg):
    if err:
        print("delivery failed:", err)
    else:
        print(f"  -> partition {msg.partition()} offset {msg.offset()}")

print("producing events to 'gpu-usage' (Ctrl-C to stop)...")
while True:
    gpu = random.choice(GPUS)
    event = {
        "gpu_id": gpu,
        "utilization_pct": random.randint(10, 100),
        # event-time: when the usage was measured, stamped by the source
        "event_time": datetime.now(timezone.utc).isoformat(),
    }
    producer.produce(
        TOPIC,
        key=gpu,                       # same key -> same partition -> ordered
        value=json.dumps(event),
        callback=delivered,
    )
    producer.poll(0)                   # let delivery callbacks fire
    print("sent:", event)
    time.sleep(1)

3. The consumer. Save as consumer.py. It joins a consumer group, reads events, and keeps a running count per GPU:

import json
from collections import Counter
from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "usage-counter",       # the consumer GROUP id
    "auto.offset.reset": "earliest",   # if no saved offset, start from the beginning
})
consumer.subscribe(["gpu-usage"])

counts = Counter()
print("consuming from 'gpu-usage' (Ctrl-C to stop)...")
try:
    while True:
        msg = consumer.poll(1.0)       # wait up to 1s for an event
        if msg is None:
            continue
        if msg.error():
            print("error:", msg.error())
            continue
        event = json.loads(msg.value())
        counts[event["gpu_id"]] += 1
        print(f"offset {msg.offset():>4} | {event['gpu_id']} "
              f"{event['utilization_pct']}%  | totals: {dict(counts)}")
finally:
    consumer.close()                   # leaves the group cleanly, commits offsets

4. Run them together. Open two terminals. Start the consumer first, then the producer:

# terminal 1
python consumer.py

# terminal 2
python producer.py

When streaming is worth it

You've now felt how clean the happy path is — so here's the honest counterweight. Streaming's real cost isn't writing the producer and consumer; it's operating them.

• It's always on. A batch job that fails at 2 a.m. waits for you. A streaming consumer that crashes is silently falling behind every second it's down — lag is a metric you must watch.
• Debugging is harder. There's no tidy input file to inspect. You reason about offsets, lag, partition assignment, and "which event triggered this bad output" across an unbounded stream.
• Correctness is subtler. At-least-once means duplicates; out-of-order arrival means late data and watermark tuning. None of this exists in a batch job over a static table.

A common, sane middle ground: stream the ingestion (so events land durably in the log within seconds), then run batch jobs that read from the log on a schedule. You get a buffered, replayable front door without the burden of real-time processing everywhere downstream.

✓ Check yourself

Answer from memory before opening each one.

1. Someone removes key=gpu from the producer, and the gpu-usage topic has three partitions. A consumer that tracks each GPU's latest utilization starts showing stale values now and then. Why, and what did the key guarantee?

   Show answer

   Without a key, one GPU's events are spread across partitions, and ordering is guaranteed only within a partition — so an older reading from one partition can be processed after a newer one from another and overwrite it. The key routes all events for the same GPU to the same partition: each GPU's history stays ordered while different GPUs still spread out for parallelism. See The log abstraction.

2. A consumer adds each event's GPU-hours to a running total in a database and then commits its offset. It crashes between those two steps. What does the total look like after the restart, and what property must the write have to make that harmless?

   Show answer

   The event is counted twice: the offset was never committed, so on restart the consumer re-reads it and adds it again. That is at-least-once delivery — you may see an event more than once but never lose one — so downstream logic must be idempotent (for example, an upsert keyed on the event rather than a blind increment). See Producers, consumers & offsets.

3. gpu-usage has three partitions. To go faster you start five consumers in the group usage-counter, and the billing team starts one consumer in a new group of their own. How many of your five do any work, and how many of the events does billing's consumer see?

   Show answer

   Three of yours work and two sit idle: within a group each partition is handed to exactly one member, so partitions cap a group's parallelism. Billing sees every event — reading doesn't remove anything from the log, and each group keeps its own offsets, so independent readers don't take events from each other. See consumer groups and Why "durable log" matters.

4. Run the exercise's consume_all.py a second time without producing anything new. What does it print, why — and what would you change to read the five events again?

   Show answer

   Nothing; it sits in its loop waiting. The first run committed the group demo-reader's offset past the fifth event, and auto.offset.reset: earliest only applies when a group has no saved offset. The events are still on disk, so replaying means moving the bookmark: use a new group.id, or reset that group's offset to 0. See offsets and replay.

5. A "GPU-hours per minute" aggregation buckets events by the time the consumer reads them. You replay yesterday's events after fixing a bug, and the per-minute numbers come out different from the first time. Why — and which clock makes the replay reproduce the same answer?

   Show answer

   Processing-time depends on when the consumer happened to run and how delayed each event was, so the same events land in different windows on every run — in a replay, a whole day may fall into a few minutes. Windowing by event-time (the timestamp stamped at the source) gives the same buckets no matter when you process; the price is handling late events, which is what a watermark and a late-data policy are for. See Event-time & windows.

6. The marketplace wants two things: GPU capacity that scales up within seconds of a utilization spike, and a weekly utilization report for the operations team. Which one justifies streaming, what question decides it, and what middle ground could serve both?

   Show answer

   Only the autoscaling: ask whether a result that is an hour old is worthless. For autoscaling it is — freshness is the product; for a weekly report a batch table is simpler, cheaper, easier to test and doesn't need to stay up 24/7. The middle ground is to stream the ingestion so events land durably in the log, let the autoscaler consume them live, and have scheduled batch jobs read from the same log for the report. See When streaming is worth it.

import json
from confluent_kafka import Producer

p = Producer({"bootstrap.servers": "localhost:9092"})
for i in range(5):
    p.produce("demo", key="g1", value=json.dumps({"n": i}))
p.flush()                              # block until all 5 are delivered
print("produced 5 events")

from confluent_kafka import Consumer

c = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "demo-reader",
    "auto.offset.reset": "earliest",   # no saved offset -> start at offset 0
})
c.subscribe(["demo"])

seen = 0
while seen < 5:
    msg = c.poll(1.0)
    if msg is None or msg.error():
        continue
    print(f"offset {msg.offset()} -> {msg.value().decode()}")
    seen += 1
c.close()

python produce_five.py
python consume_all.py        # prints offsets 0,1,2,3,4

Takeaway

• Batch buys simplicity and streaming buys freshness at the price of always-on infrastructure, harder debugging and subtler correctness; most analytics belongs in batch.
• A topic is an append-only log that keeps events until retention expires, so reading does not remove anything, and replay and multiple independent readers follow from that.
• Ordering holds only within one partition, so key events by the entity whose order matters, such as the GPU id.
• An offset is each consumer's own bookmark; committing it after processing gives at-least-once delivery, so consumers must tolerate duplicates.
• A consumer group gives each partition to one member, which is how you scale reads without double-processing.
• Windows should use event time, not processing time, and a watermark decides when a window is safe to close; closing early drops late events and waiting forever emits nothing.
• Consumer lag is the metric to watch, since a crashed consumer falls further behind every second it is down.

Most of the pipeline still runs as scheduled jobs that depend on each other, and something has to order them and recover from failure. → Orchestration with Dagster

Further reading

• Apache Kafka: Concepts and terms — Apache Kafka docs, accessed 2026. Topics, partitions, offsets and consumer groups as the Kafka API defines them.
• Apache Kafka: Design — Apache Kafka docs, accessed 2026. Why the log is built this way, including delivery semantics.
• Redpanda: Introduction to Redpanda — Redpanda docs, accessed 2026. The same model on the engine used in the lab.
• The world beyond batch: Streaming 101 — Akidau, O'Reilly Radar, 2015. The clearest account of event time, processing time and windows.
• The Dataflow Model — Akidau et al., VLDB, 2015. The paper behind watermarks and the correctness, latency and cost trade-off.