Lesson 9 · Transform, warehouse & the whole pipeline

ksqlDB, stream processing

For the warehouse path, raw change topics aren't the final shape — ksqlDB joins and reshapes them into query-friendly dimensions, in SQL, as they flow.

Your win: explain the difference between a ksqlDB stream and a table, and read how this repo turns raw users change events into an enriched dim_user shape.

The transform stage — only on the warehouse path

Remember from Lesson 1: plain cross-service sync goes source → topic → sink, no transform. But the warehouse path inserts a step: ksqlDB, a stream-processing engine that runs SQL over Kafka topics — joining, aggregating, and reshaping change streams into the dim_*/fact_* tables analytics wants.1

Streams vs tables — the one idea

Stream

An unbounded, append-only series of events — every change, in order. A change topic read as "everything that happened." CREATE STREAM … WITH (VALUE_FORMAT='AVRO').

Table

The latest value per key — a materialized view of "current state." Built by collapsing a stream: for each key, keep the most recent value.

Both are built from the same Kafka topic — the difference is interpretation: a stream is the history, a table is the now.2 (Queries come in two flavours too: push queries subscribe to results continuously — EMIT CHANGES; pull queries fetch the current value once, like a normal DB query.)

How dim_user is built

Watch the transform assemble, from real ksqlDB SQL. First, a stream over the raw CDC topic; then a table that keeps the latest value per user:

deployments/helm/backend/hephaestus/datawarehouses/lms/ksql/migrations/V000002__dim_user.sql:7-47 (condensed)
CREATE STREAM BOB_USERS_STREAM_ORIGIN_V1
    WITH (KAFKA_TOPIC='…bob.public.users', VALUE_FORMAT='AVRO');   -- the raw change stream

CREATE TABLE BOB__USERS__V8 AS SELECT
    AFTER->USER_ID AS USER_ID,                          -- read the envelope's "after" row
    LATEST_BY_OFFSET(AFTER->NAME)  AS NAME,
    LATEST_BY_OFFSET(AFTER->EMAIL) AS EMAIL, -- …
  FROM BOB_USERS_STREAM_ORIGIN_V1
  GROUP BY AFTER->USER_ID                              -- one row per user = current state
  EMIT CHANGES;                                       -- push query: keep it updated forever

LATEST_BY_OFFSET is how a stream becomes a table: for each USER_ID, keep the most recent value of each column. (Note ksqlDB reads AFTER->… directly — it has the Debezium envelope right there, so it doesn't need the sink's unwrap SMT.) Later in the same file, that table is joined with user-groups (users ⋈ user_group_member ⋈ user_group) to enrich each user with their group, producing the final dim_user shape.

ksqlDB owns its own sink connectors

The tie back to hephaestus The ksqlDB SQL doesn't stop at building the table — it also declares the sink that writes it to the warehouse: CREATE SINK CONNECTOR … JdbcSinkConnector … table.name.format = public.dim_user. So ksqlDB creates and owns those sink connectors, not the generation system. That's exactly why hephaestus's delete-guard (Lesson 8) skips all-caps connector names — those are ksqlDB-managed, and the control plane must not touch them. The naming convention is the ownership boundary.
"ksqlDB migrations" The transforms live in versioned SQL files (V000002__dim_user.sql, …) applied by Confluent's ksql-migrations CLI — a schema-migration tool for ksqlDB, just like golang-migrate for Postgres. hephaestus only seeds the bookkeeping it needs: a MIGRATION_EVENTS stream and a schema-versions table (ksqldb/init_migrate.go). The actual migrations run in the deploy hook (5_migrate_ksql, Lesson 8).
Read this next

ksqlDB streams & tables

The two first-class types and the stream/table duality — the mental model this transform rests on — plus push vs pull queries.

→ Confluent — streams & tables
→ ksqlDB — queries (push vs pull) · in-repo …/ksql/migrations/V000002__dim_user.sql

Check yourself (from memory)

Q1. What's the difference between a ksqlDB stream and a table?

Same topic, two interpretations: the stream is "everything that happened," the table is "current state per key" (via LATEST_BY_OFFSET).

Q2. Which connectors does ksqlDB itself create and own?

CREATE SINK CONNECTOR in the ksqlDB SQL — which is why the delete-guard skips all-caps names.

Q3. On which path does the ksqlDB transform run?

ksqlDB is the transform stage for the warehouse stacks; direct data-sync goes source → topic → sink with no ksqlDB.
Recall: ksqlDB — streams, tables, and how dim_user is built.
stream/table + the build + ownership, then reveal
ksqlDB = SQL stream processing over Kafka; the transform stage on the warehouse path (plain sync skips it). Stream = append-only event history; table = latest value per key (materialized view) — same topic, two interpretations. Queries: push (EMIT CHANGES, continuous) vs pull (point-in-time). Build dim_user: CREATE STREAM … VALUE_FORMAT='AVRO' from the CDC topic → CREATE TABLE AS SELECT AFTER->col, LATEST_BY_OFFSET(...) GROUP BY key EMIT CHANGES (stream→table = current state) → JOIN users ⋈ user_group_member ⋈ user_group → enriched dim. ksqlDB reads AFTER-> directly (no unwrap SMT). ksqlDB OWNS its sink connectors (CREATE SINK CONNECTOR in the SQL; all-caps names → hephaestus's delete-guard skips them). Transforms = versioned SQL via Confluent ksql-migrations.
🎯 Interview one-liner "How do you reshape CDC data for analytics?" → "ksqlDB. It reads the raw change topics as streams, collapses them to tables with LATEST_BY_OFFSET for current state, joins across entities to enrich them, and writes the result as dimension tables — all in SQL that runs continuously. ksqlDB even declares its own sink connectors, so those are managed separately from our generated ones."
Next: where those dim_* tables land — the data warehouse, its star schema, and the chained CDC that feeds partner warehouses. Ask me about the stream/table duality if it's slippery — it's the heart of stream processing.

1. Confluent — ksqlDB streams & tables. In-repo: …/ksql/migrations/V000002__dim_user.sql:7-47.

2. ksqlDB — queries. In-repo: internal/platform/ksqldb/init_migrate.go.