Lab 04 — The Telemetry Firehose

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 of active rentals. In the real platform this list
# would come from the rentals table; for the lab a fixed set is plenty.
ACTIVE_GPUS = [f"gpu-{i:03d}" for i in range(1, 21)]   # 20 cards


def reading(gpu_id: str) -> 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=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()

import json
import duckdb
from confluent_kafka import Consumer

TOPIC = "telemetry"
BROKER = "localhost:19092"
DB = "warehouse.duckdb"
BATCH = 200          # land in chunks of this many readings

DDL = """
CREATE SCHEMA IF NOT EXISTS bronze;
CREATE TABLE IF NOT EXISTS bronze.telemetry (
    gpu_id     VARCHAR,
    ts         TIMESTAMP,
    util_pct   DOUBLE,
    temp_c     DOUBLE,
    power_w    DOUBLE,
    loaded_at  TIMESTAMP DEFAULT now()
);
"""


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 = []
    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
                continue
            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:
        flush(con, buffer)
        total = con.execute("SELECT count(*) FROM bronze.telemetry").fetchone()[0]
        print(f"\nStopped. bronze.telemetry now holds {total} readings.")
    finally:
        consumer.close()
        con.close()


if __name__ == "__main__":
    main()

{{ 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.gpu_model,
    g.region,
    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

version: 2
sources:
  - name: bronze
    schema: bronze
    tables:
      - name: telemetry

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.gpu_model,
    g.region,
    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.gpu_model, g.region;

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 20 readings (total N)" every 5 seconds — let it run ~3–5 min

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

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, avg_util_pct, max_temp_c, sample_count
  FROM gold.fct_telemetry_hourly
  ORDER BY hour, gpu_id LIMIT 8;"

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