Lab 04 — The Telemetry Firehose

The goal

So far your data has been batchy: rentals and customers extracted from Postgres, modeled into a star schema. Telemetry is different. A single GPU under load emits a metric reading every few seconds — and the reference marketplace has thousands of cards running at once. If you tried to load every raw reading into the warehouse and let the BI tool query it, you'd melt your laptop and your bill. The pattern that saves you is the dual path:

By the end of this lab you'll produce a live stream of per-GPU metrics, consume and batch-land them into a bronze telemetry table, and build fct_telemetry_hourly — a gold fact with one row per GPU per hour (average utilization, peak temperature, sample count), joined to dim_gpu. The dashboard in Lab 08 will read that fact, never the raw firehose.

Produce the telemetry stream

A telemetry event is small and regular. We'll emit one JSON message per active rental every few seconds, with this shape:

Create a small producer beside the generator. It reads the set of currently-active GPUs (here we simulate a fixed fleet: one card per catalog GPU from Lab 01) and pushes a reading for each, on a loop. Notice we key each message by gpu_id — that keeps all of one card's readings on the same partition, which matters once you scale to multiple consumers.

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

TOPIC = "telemetry"
BROKER = "localhost:19092"   # Redpanda's external listener from the Lab 01 compose

# A small simulated fleet: one active card per catalog GPU, keyed by the
# Lab 01 gpu_id (1-10) so readings join dim_gpu. In the real platform this
# list would come from the rentals table; for the lab a fixed set is plenty.
ACTIVE_GPUS = list(range(1, 11))   # 10 cards


def reading(gpu_id: int) -> dict:
    """One realistic-ish telemetry sample for a GPU under load."""
    util = random.gauss(78, 12)                 # mostly busy, some idle dips
    util = max(0.0, min(100.0, util))
    return {
        "gpu_id": gpu_id,
        "ts": datetime.now(timezone.utc).isoformat(),
        "util_pct": round(util, 1),
        "temp_c": round(45 + util * 0.45 + random.gauss(0, 2), 1),   # hotter when busier
        "power_w": round(120 + util * 2.6 + random.gauss(0, 8), 1),  # more draw when busier
    }


def main(interval_s: float = 5.0):
    producer = Producer({"bootstrap.servers": BROKER})
    print(f"Producing telemetry for {len(ACTIVE_GPUS)} GPUs every {interval_s}s → topic '{TOPIC}'. Ctrl-C to stop.")
    sent = 0
    try:
        while True:
            for gpu_id in ACTIVE_GPUS:
                event = reading(gpu_id)
                producer.produce(TOPIC, key=str(gpu_id), value=json.dumps(event))
                sent += 1
            producer.flush()             # push this batch to Redpanda
            print(f"  sent {len(ACTIVE_GPUS)} readings (total {sent})")
            time.sleep(interval_s)
    except KeyboardInterrupt:
        producer.flush()
        print(f"\nStopped. {sent} readings produced.")


if __name__ == "__main__":
    main()

Consume & land to bronze

The consumer is the other half of the stream path. It subscribes to the topic, pulls messages in batches, and writes them into a bronze.telemetry table in DuckDB. Two things make this production-shaped:

• Batch the writes. Inserting one row per message is slow. We buffer a few hundred readings, then do a single bulk insert — far fewer, fatter writes.
• Land it raw and untouched. Bronze keeps the data exactly as it arrived (plus a load timestamp). No cleaning, no joins — that's the next stage's job. Raw stays cheap and to the side.

import json
import sys
import duckdb
from confluent_kafka import Consumer

TOPIC = "telemetry"
BROKER = "localhost:19092"
DB = "warehouse.duckdb"
BATCH = 200          # land in chunks of this many readings
DRAIN = "--drain" in sys.argv   # land what is waiting, then exit once the topic is idle
IDLE_POLLS_TO_EXIT = 5          # ...idle meaning this many empty 1s polls in a row

DDL = """
CREATE SCHEMA IF NOT EXISTS bronze;
CREATE TABLE IF NOT EXISTS bronze.telemetry (
    gpu_id     INTEGER,
    ts         TIMESTAMPTZ,                 -- an instant, not a wall-clock reading
    util_pct   DOUBLE,
    temp_c     DOUBLE,
    power_w    DOUBLE,
    loaded_at  TIMESTAMPTZ DEFAULT now()    -- when the row landed (Lab 06 reads this)
);
"""


def flush(con, rows):
    """Bulk-insert a buffer of readings into bronze, then clear it."""
    if not rows:
        return
    con.executemany(
        "INSERT INTO bronze.telemetry (gpu_id, ts, util_pct, temp_c, power_w) "
        "VALUES (?, ?, ?, ?, ?)",
        rows,
    )
    print(f"  landed {len(rows)} readings → bronze.telemetry")
    rows.clear()


def main():
    con = duckdb.connect(DB)
    con.execute(DDL)

    consumer = Consumer({
        "bootstrap.servers": BROKER,
        "group.id": "telemetry-lander",
        "auto.offset.reset": "earliest",   # read from the start the first time
    })
    consumer.subscribe([TOPIC])
    print(f"Consuming '{TOPIC}' → {DB} (batch={BATCH}). Ctrl-C to stop.")

    buffer = []
    idle_polls = 0
    try:
        while True:
            msg = consumer.poll(1.0)        # wait up to 1s for a message
            if msg is None:
                flush(con, buffer)          # idle → flush whatever we have
                idle_polls += 1
                if DRAIN and idle_polls >= IDLE_POLLS_TO_EXIT:
                    break                   # drained: nothing new for a few seconds
                continue
            idle_polls = 0
            if msg.error():
                print("  consumer error:", msg.error())
                continue
            e = json.loads(msg.value())
            buffer.append((e["gpu_id"], e["ts"], e["util_pct"], e["temp_c"], e["power_w"]))
            if len(buffer) >= BATCH:
                flush(con, buffer)
    except KeyboardInterrupt:
        pass
    finally:
        flush(con, buffer)
        total = con.execute("SELECT count(*) FROM bronze.telemetry").fetchone()[0]
        print(f"\nStopped. bronze.telemetry now holds {total} readings.")
        consumer.close()
        con.close()


if __name__ == "__main__":
    main()

Roll up to an hourly fact

Here's the payoff. Bronze now holds hundreds of tiny raw readings (thousands if you let it run) — and we are never going to query them from the warehouse. Instead we collapse them into one row per GPU per hour. That's a 700-to-1 reduction (a card sampled every 5s = 720 readings/hour → 1 rollup row), and it's the only telemetry the rest of the platform ever sees.

Add this as a dbt model in your gold marts (the dbt project from Lab 03). It buckets readings into hourly windows, aggregates them, and joins dim_gpu so business attributes ride along:

{{ config(materialized='table') }}

with readings as (
    select
        gpu_id,
        date_trunc('hour', ts) as hour,
        util_pct,
        temp_c,
        power_w
    from {{ source('bronze', 'telemetry') }}
),

rolled as (
    select
        gpu_id,
        hour,
        round(avg(util_pct), 1)  as avg_util_pct,
        max(temp_c)              as max_temp_c,
        round(avg(power_w), 1)   as avg_power_w,
        count(*)                 as sample_count
    from readings
    group by gpu_id, hour
)

select
    r.gpu_id,
    r.hour,
    g.model,
    r.avg_util_pct,
    r.max_temp_c,
    r.avg_power_w,
    r.sample_count
from rolled r
left join {{ ref('dim_gpu') }} g on r.gpu_id = g.gpu_id

You'll need to register the new bronze table so source('bronze', 'telemetry') resolves. Add one line to the Lab 03 sources file (a second file declaring a source called bronze would make dbt stop with a duplicate-source error):

    tables:
      - name: customers
      - name: gpus
      - name: rentals
      - name: gpu_prices
      - name: telemetry     # landed by the Lab 04 stream consumer

Prefer raw SQL while you're learning the shape? The same rollup, runnable directly against DuckDB, looks like this — useful for sanity-checking before the dbt run:

CREATE SCHEMA IF NOT EXISTS gold;
CREATE OR REPLACE TABLE gold.fct_telemetry_hourly AS
SELECT
    t.gpu_id,
    date_trunc('hour', t.ts)        AS hour,
    g.model,
    round(avg(t.util_pct), 1)       AS avg_util_pct,
    max(t.temp_c)                   AS max_temp_c,
    round(avg(t.power_w), 1)        AS avg_power_w,
    count(*)                        AS sample_count
FROM bronze.telemetry t
LEFT JOIN gold.dim_gpu g ON t.gpu_id = g.gpu_id
GROUP BY t.gpu_id, date_trunc('hour', t.ts), g.model;

Run it end to end

You'll run the producer and consumer at the same time, in two terminals. Start the consumer first so it's listening, then the producer, and let them run for a few minutes to accumulate a meaningful sample.

cd mini-griddp
uv run python ingest/consume_stream.py
# leave this running; it prints "landed N readings" as batches arrive

cd mini-griddp
uv run python generator/telemetry_producer.py
# prints "sent 10 readings (total N)" every 5 seconds — let it run ~3–5 min

Now build the rollup. With dbt:

cd mini-griddp/dbt
uv run dbt run --select fct_telemetry_hourly

Then inspect what you built:

duckdb ../warehouse.duckdb -c "
  SELECT count(*) AS raw_readings FROM bronze.telemetry;
  SELECT count(*) AS rollup_rows  FROM gold.fct_telemetry_hourly;
  SELECT gpu_id, hour, model, avg_util_pct, max_temp_c, sample_count
  FROM gold.fct_telemetry_hourly
  ORDER BY hour, gpu_id LIMIT 8;"

What this looks like at real scale

Commit Lab 04

You've added the streaming path and the firehose rollup to the platform. Save it.

cd mini-griddp
git add generator/telemetry_producer.py ingest/consume_stream.py \
        dbt/models/marts/fct_telemetry_hourly.sql dbt/models/staging/_sources.yml
git commit -m "Lab 04: telemetry firehose — produce, land to bronze, roll up to fct_telemetry_hourly"

✓ Check yourself

Answer from memory before opening each one.

1. An analyst asks for a Metabase chart straight off bronze.telemetry so they can see every reading. What breaks first as the fleet grows, and what do you offer instead?

   Show answer

   Every dashboard load would scan the raw firehose — 720 readings per GPU per hour at a 5-second interval, on the order of 150 million a day at the reference scale — so query time and cost grow with the fleet. Raw stays landed to one side; dashboards read fct_telemetry_hourly, where those 720 readings are one row. See the dual path and the rollup.

2. You run the producer (10 GPUs, one reading each every 5 seconds) for four minutes, all inside one clock hour. Roughly how many rows should bronze.telemetry and fct_telemetry_hourly hold — and what changes if the four minutes straddle the top of the hour?

   Show answer

   About 10 × 12 × 4 = 480 raw readings and 10 rollup rows, each with a sample_count near 48. Straddle the hour and you get 20 rollup rows: date_trunc('hour', ts) buckets by clock hour, not by elapsed time, so each GPU's samples split across two rows. See Roll up to an hourly fact and Run it end to end.

3. Every message is produced with key=gpu_id. With a 3-partition topic and three consumers, what does that key guarantee — and what does it not?

   Show answer

   The same key always goes to the same partition, so all of one card's readings stay in order and are handled by a single consumer. It does not give each GPU a partition of its own: ten keys are hashed across three partitions, so every partition carries several cards. See Produce the telemetry stream.

4. You rename the consumer's group.id from telemetry-lander to telemetry-lander-v2 and restart it. What lands in bronze.telemetry?

   Show answer

   Everything still held in the topic, all over again. Redpanda remembers the read position per group; a new group has none, and auto.offset.reset: earliest starts it from the beginning, so each retained reading is inserted a second time and every sample_count doubles at the next rollup. The group id is the thing that lets a restart resume. See Why a consumer group?

5. For one GPU, hour A has 700 samples averaging 90% utilization and hour B has 20 samples averaging 10%. A daily rollup built on the hourly fact takes avg(avg_util_pct). What does it report, what is the true figure, and which hourly column rescues you?

   Show answer

   It reports 50%; the true mean is (700 × 90 + 20 × 10) ÷ 720 ≈ 87.8%. An average of averages gives the 20-sample hour the same weight as the 700-sample one, so weight each hour by sample_count — which is why the rollup keeps it. max(max_temp_c) needs no such care: maxima compose, averages do not. See the rollup's columns and the exercise below.

6. Ops wants to know whenever a card runs above 90 °C for more than two minutes. How far does the hourly rollup get them, and what decides whether you build something finer?

   Show answer

   max_temp_c shows that a card peaked above 90 at some point in the hour — not when, nor for how long — and because the rollup is built in a batch, it arrives too late to act on. If the real question is a trend ("which cards run hot?"), hourly is enough and the cheapest thing to serve. If duration or a fast reaction matters, you need a finer bucket — a per-minute rollup, or a check on the stream itself — at the price of 60× the rows. Same pattern, different grain. See the rollup and real scale.

{{ config(materialized='table') }}

select
    gpu_id,
    model,
    cast(hour as date)                          as day,
    -- weight each hour's average by its sample_count for a true daily mean
    round(sum(avg_util_pct * sample_count)
          / nullif(sum(sample_count), 0), 1)    as avg_util_pct,
    max(max_temp_c)                             as max_temp_c,
    sum(sample_count)                           as sample_count
from {{ ref('fct_telemetry_hourly') }}
group by gpu_id, model, cast(hour as date)

cd mini-griddp/dbt
uv run dbt run --select fct_telemetry_daily
duckdb ../warehouse.duckdb -c "SELECT * FROM gold.fct_telemetry_daily ORDER BY gpu_id LIMIT 5;"

Takeaway

• High-volume telemetry is landed raw to the side, never joined to dimensions and never queried by a dashboard, and only the rollup enters the modeled warehouse.
• The rollup is one row per GPU per hour, which cuts the volume by orders of magnitude and keeps business joins cheap.
• Key each message by gpu_id so one card's readings stay on one partition and in order, which matters once you add consumers.
• The consumer buffers readings and writes them in batches, because one insert per message is slow.
• A consumer group id makes Redpanda remember the consumer's offset, so a restart resumes where it stopped instead of landing everything again.
• The rollup row count tracks the hours you ran: ten GPUs over one hour give ten rows, and twenty if the run crossed the top of an hour.
• The production shape is the same as the lab, land raw, aggregate into time buckets, serve the rollups, and only the engines and tiers change.

You now have ingest, models and a rollup, all run by hand, so the next lab makes the platform run itself as a scheduled graph. → Lab 05 — Orchestrate with Dagster

Further reading

• Redpanda: Consumer offsets — Redpanda docs, accessed 2026. How group offsets are committed and resumed.
• Redpanda: Configure producers — Redpanda docs, accessed 2026. Keys, batching and acknowledgements on the producer side.
• Redpanda: rpk topic create — Redpanda docs, accessed 2026. Creating a topic with an explicit partition count.
• Apache Kafka: Design — Apache Kafka docs, accessed 2026. Why ordering holds per partition and what a key does.