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

• Can you explain the difference between batch and streaming in one sentence each — and say which one most pipelines should use?
• Why is ordering only guaranteed within a partition, not across a topic?
• What does an offset bookmark let you do that a delete-on-read queue cannot?
• Why bucket aggregations by event-time instead of processing-time?

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

Next

You can move data event-by-event — but most of your pipeline still runs as scheduled jobs that depend on each other, and something has to run them in the right order and recover when they fail. → Orchestration with Dagster