Sensor Stream

A deliberately over-engineered toy pipeline for learning data engineering. It takes three fake temperature sensors and pushes them through a full streaming lakehouse — far more machinery than the problem needs, which is the point.

Goals

• Medallion architecture — data flows through refining layers:
  • Bronze — raw events, exactly as received.
  • Silver — cleaned (all temperatures → Celsius) and enriched (city/country joined in from Postgres).
  • Gold — business aggregates (hourly avg/min/max temperature per city).
• Separate storage from compute — the data lives in object storage (MinIO) as open files; compute engines are separate and swappable. Stop DuckDB and the Bronze, Silver, and Gold data remains available in MinIO.

Architecture

                          STORAGE (always on)
 generator ─▶ Kafka ─▶ Flink ─────────────▶ MinIO (s3://sensor-data)
 (Python)             (stream)               ├─ bronze/   raw JSON
                        ▲                    ├─ silver/   Celsius + location
                        │ lookup             └─ gold/     hourly aggregates (Parquet)
                     Postgres                      ▲
                    (locations)                    │
                                              DuckDB (stateless compute)
                                             materializes gold from silver

• Flink streams Kafka → Bronze + Silver (one job, two sinks).
• DuckDB is the stateless batch/query compute engine. A scheduled loop reads Silver from MinIO and writes externally materialized Gold aggregates back as Parquet. The interactive shell queries Bronze, Silver, and Gold directly from MinIO. No persistent DuckDB database file is used; MinIO owns the durable data.

Gold is therefore not a DuckDB-managed materialized view. The refresh loop recomputes it explicitly, and querying Gold reads the resulting Parquet files from MinIO.

Usage

./start.sh             # build + run everything (bronze, silver, and auto-refreshing gold)
./query.sh             # open an interactive DuckDB SQL shell over the lake
./correct-location.sh  # reprocessing demo: fix a sensor's location, replay history
./stop.sh              # stop containers
./reset.sh             # stop and wipe all data

./start.sh waits for infrastructure health checks, Flink's first finalized bronze and silver files, and DuckDB's first gold refresh. A clean first start normally becomes queryable in about 60–90 seconds after image builds finish, with a two-minute timeout. When it prints Sensor Stream is queryable., ./query.sh can open all three views without its own readiness wait. Restarts with existing lake data are typically faster.

Restarts resume where they left off. Kafka's log and the Flink source's committed offsets are persisted, so ./stop.sh && ./start.sh continues the stream instead of replaying the whole topic into bronze/silver. (It's resume, not exactly-once — the file sinks aren't transactional, so a hard crash could still dup/drop a handful of rows at the restart boundary.) ./reset.sh wipes everything, including Kafka, for a clean slate.

In the query shell, bronze, silver, and gold are predefined views. Start with the raw events exactly as Flink received them:

SELECT sensor_id, temperature, unit, event_time
FROM bronze
ORDER BY event_time DESC
LIMIT 8;

Silver remains event-level, but normalizes every temperature to Celsius and enriches each reading with its location:

SELECT strftime(event_time::TIMESTAMPTZ, '%Y-%m-%d %H:%M:%S') AS event_time,
       sensor_id,
       round(temperature_celsius, 2) AS temp_c,
       city
FROM silver
ORDER BY event_time::TIMESTAMPTZ DESC
LIMIT 7;

Gold changes the grain: many silver events become one hourly aggregate per city:

SELECT city,
       strftime(hour, '%Y-%m-%d %H:00') AS hour,
       readings,
       avg_temp_c AS avg_c,
       min_temp_c AS min_c,
       max_temp_c AS max_c
FROM gold
ORDER BY hour DESC, city
LIMIT 6;

More queries to try:

-- What raw units arrived?
SELECT unit, count(*) AS readings
FROM bronze
GROUP BY unit;

-- Temperature range by city, computed from event-level silver.
SELECT city,
       round(min(temperature_celsius), 2) AS min_c,
       round(avg(temperature_celsius), 2) AS avg_c,
       round(max(temperature_celsius), 2) AS max_c
FROM silver
GROUP BY city
ORDER BY city;

-- Follow one city's hourly trend from the aggregate layer.
SELECT hour, readings, avg_temp_c
FROM gold
WHERE city = 'New York'
ORDER BY hour DESC
LIMIT 12;

Helsinki appears as a city after running the location-correction example below; a fresh stack intentionally labels hel-001 as Espoo.

Reprocessing (a.k.a. "backfill")

Reprocessing rebuilds silver from bronze — the durable raw layer is replayed through the silver transform (see duckdb/backfill.sql), independent of Kafka retention. Each rebuilt row gets a _rebuilt_at stamp so you can see it land.

You reprocess when the raw data is fine but something downstream was wrong: a transform bug, a new silver column, or a correction to reference data. (If a sensor itself was broken, there's nothing to reprocess — the readings were never captured. That's a data gap, not a backfill.)

Worked example — a mislabeled sensor. hel-001 is recorded as Espoo in the Postgres location table, but it's actually in Helsinki. Silver is enriched at stream time, so every historical silver row already says "Espoo". Correcting Postgres alone only fixes rows streamed from that point on — history stays wrong until you replay it. ./correct-location.sh does the whole loop: show silver (all Espoo) → correct the reference → stop the streaming job → clear + replay bronze → silver → rebuild gold → submit a fresh streaming job → show silver again (now all Helsinki).

./correct-location.sh
SELECT sensor_id, city, count(*), max(_rebuilt_at) FROM silver GROUP BY 1, 2;

This is the payoff of an immutable bronze layer: because the raw history is kept intact and complete, any derived layer can be rebuilt from it whenever the logic or reference data changes.

For a different reprocessing reason, such as a transform change in duckdb/backfill.sql, use the lifecycle in correct-location.sh as the template: stop active Flink jobs, replace silver, rebuild gold, and submit one fresh pipeline job. Do not delete silver while Flink is running: its checkpoint may still reference in-progress files, causing recovery failures.

UIs

Service	URL
Flink	http://localhost:8081
MinIO	http://localhost:9001
Kafka UI	http://localhost:8080

Simplifications

This is a learning demo — it deliberately cuts corners to keep the concepts front and center:

• Restarts resume but aren't exactly-once — the file sinks aren't transactional, so a hard crash can dup/drop a few rows at the boundary.
• Silver/gold are plain JSON/Parquet files, so backfill overwrites a layer wholesale (a real setup would partition by date or use a table format like Iceberg for partition-scoped, atomic, versioned rewrites).
• Gold is recomputed in full each cycle rather than incrementally.
• Credentials are hard-coded and everything runs single-node.