Streaming#

An optional event stream beside the lake (plan2 D18 part B): Apache Kafka, a schema registry, and a sink service that lands each stream in the lake as a Delta table, exactly once, with the same tags and audit columns as every other table. Dagster observes the streams but doesn’t process them.

Status: the services, the sink, one demo stream, streams in the catalog and a streaming leg in the round-trip proof are built. Still to come: publishing lake tables as streams, and the AWS side (MSK).

Turning it on#

Streaming is off by default. Set the Compose profile in your core .env and bring the stack up:

COMPOSE_PROFILES=streaming

The profile adds four services to the core stack, reachable only on the stack’s own network. Kafka doesn’t create topics on its own (KAFKA_AUTO_CREATE_TOPICS_ENABLE=false): the platform creates each stream’s topic with its partitions, so a stray client can’t create one with the broker’s default.

Service	What it is	Memory limit (default)
`kafka`	Apache Kafka 4.3, one node in KRaft mode (broker and controller in one process); data on its own volume	1 GB (`ODS_KAFKA_MEMORY`; heap `ODS_KAFKA_HEAP`, 512 MB)
`schema-registry`	Apicurio Registry 3.3, in its own `schema_registry` database on the stack’s Postgres; clients use its Confluent-compatible API	768 MB (`ODS_SCHEMA_REGISTRY_MEMORY`)
`schema-registry-init`	creates that database once, then exits
`stream-sink`	lands the demo stream (below)	512 MB (`ODS_STREAM_SINK_MEMORY`)

Idle, Kafka and the registry use about 0.6 GB together. With the profile on, Dagster also loads the stream assets, the demo job and the observer (it reads ODS_STREAMING, set from COMPOSE_PROFILES).

A stream is declared by its schema#

A stream’s Avro schema is the declaration of both the events and the table they land in, the way an asset’s declared columns are for a table (asset metadata). Each field’s doc becomes the column description, and the column tags travel inside the schema as field attributes, which Avro lets any schema carry and requires every implementation to keep:

{"name": "ship_to_city", "type": ["null", "string"], "default": null,
 "doc": "City the order ships to.", "ods.pii": "quasi"}

The attribute names (ods.sensitivity, ods.pii, ods.key) are the ones written into the Delta files, so a tag reads the same in the registry, in the files and in the catalog. The producer registers the schema under the subject <topic>-value.

A stream is declared in code as a Stream (dagster_platform/stream/streams.py): its topic, its schema file, and the same owner, domain and sensitivity as any raw table. It lands in raw/<source>__<entity>.

The landed table#

Every event becomes one row, with the schema’s fields and then the columns the platform adds:

Column	Means
`_kafka_partition`, `_kafka_offset`	the event’s place in the log; together, the table’s key
`_kafka_timestamp`	when the broker appended the event: its ingest time, UTC
`_ods_created_at`, `_ods_modified_at`	when the sink wrote the row (audit columns)
`_ods_run_id`	`stream:<consumer group>`: no Dagster run writes these rows

The event’s own time (when it happened, by the source’s clock) is a field of the schema; in the demo, event_time.

A changed doc or tag in the stream’s schema reaches the table when the sink next starts: it updates the column’s description and tags in a metadata-only commit before landing more events (Delta won’t append data whose column metadata differs from the table’s). Delta merges column metadata, so a tag removed from the schema stays on the table, the safe side for a sensitivity or PII tag; rewrite the table to drop it.

The sink reads events with the stream’s schema as the reader schema, so a producer can add an optional field without breaking the sink; the table keeps its declared columns until the stream’s schema changes.

Exactly once#

The sink appends micro-batches (ODS_SINK_BATCH_SECONDS, default 30). Each Delta commit also records the last offset written for each partition, as a Delta application transaction (ods-sink/<topic>/<partition>). When the sink starts, it resumes from the table’s own record, not from Kafka’s committed offsets, so a crash between the Delta commit and the Kafka commit neither loses nor repeats a row. It commits Kafka’s offsets too, after each write, so the consumer lag shows in Kafka’s own tools.

A producer that sends the same event twice puts two events in the log, and both land. Deduplicating by an event id is a staging model’s job, as for any raw table.

What Dagster sees#

Writes: the sink reports each write to Dagster as a materialization of the landed table, with the usual write metadata (row count, column profile, commit history), so lineage, the catalog and freshness checks treat it like any table.
The stream: the stream_observer sensor records an observation of each stream’s table every minute: events in the topic, events landed, the lag between them, per-partition offsets and the schema version in the registry.
The catalog: a stream’s table is marked stream and has a Stream section: the topic, the schema’s subject and version, events in the topic, landed and waiting to land, and each partition’s offsets, from the latest observation.
What reads it: a derived table that reads a stream rebuilds at a pace (hourly unless changed in the portal’s settings), not after every sink write (automation).
The proof: where streaming is on, the round-trip proof also sends its generated rows through a topic of their own and the sink, and checks they land identical (stream_leg).

The demo stream#

ods.demo.order_events carries what happens to the demo orders: placed on the order date, then shipped, delivered and, for some, returned on later days. It has three partitions, keyed by order, so an order’s events stay in order. The job demo_upstream_emit_order_events plays the upstream system and publishes one day’s events (day, an ISO date). The events are seeded from the demo orders, so a day’s events are the same every time. The sink lands them in raw/demo__order_events, and the dbt model staging/stg_demo__order_events keeps one row per event (a producer may send one twice). Models that read a stream are tagged stream in dbt, so they exist only where streaming is on.

Adding a stream#

Write its Avro schema next to the code that declares it, with a doc on every field and the column tags as ods.* attributes.
Declare a Stream for it and add it to the streams the platform lands (dagster_platform/stream/defined.py) and to the definitions.
Add a sink service for it to the streaming profile: the stream-sink service with the stream’s topic as its argument.