Joins in Stream Processing

How a stream processor joins data that never stops arriving: two event streams within a time window, a stream against a table kept up to date from its changelog, and two tables into a live view.

Overview

A batch join sees both inputs complete before it starts. A stream join never does. Events keep arriving, they can arrive late or out of order, and the other side of the join may not have arrived yet. So a stream processor must remember what it has seen (its state) and decide how long to keep it. Those two questions, what to remember and for how long, shape every kind of stream join.

Stream-stream (window) join — Events — Events — Unmatched events from both sides, for the window — Match each payment to its order within 1 hour
Stream-table join (enrichment) — Events — A table, kept current from its changelog — A copy of the table — Add the customer's country to each purchase
Table-table join (view maintenance) — A table's changelog — A table's changelog — Current rows of both tables — Keep a live 'orders with customer name' view

Stream-stream join: match events within a window

Two event streams share a key, and you want to pair events that belong together: an ad impression with the click it caused, an order with its payment. Because either event can come first, the processor keeps each unmatched event in state. When the partner arrives, it emits the pair. The window bounds how long it waits: an interval join only pairs events whose timestamps are close enough.

Flink SQL expresses this with a time condition on both sides. Its docs explain that, because the event times only move forward, Flink can remove old rows from state without changing the result.

Kafka Streams makes the same choice mandatory: a KStream-KStream join is always windowed, because otherwise its state store would grow forever. The cost of a window is memory: state grows with the event rate times the window length.

The same mechanics fit in a few lines of Python. A watermark (explained below) tells the operator when it is safe to forget.

Stream-table join: enrich each event

Often one side is not a stream of facts but reference data: customer profiles, product prices, exchange rates. For each event you want to look up the matching row and add its fields. Querying the database for every event is slow and loads the database. Instead, the stream processor keeps its own copy of the table and keeps it current by reading the table's changelog, usually from change data capture (CDC). Debezium's PostgreSQL connector, for example, takes a consistent snapshot of the tables and then streams every committed row change, read from the write-ahead log, into Kafka topics.

In Kafka Streams this is a KStream-KTable join. Only the stream side triggers output: a new purchase produces an enriched purchase, while a customer update only changes the table for later purchases. The inputs must be co-partitioned (same number of partitions, same partitioning of keys), so each instance holds the part of the table it needs. A GlobalKTable avoids that requirement by giving every instance a full copy of the table.

A cheaper variant is the lookup join: Flink queries an external system, such as a database through the JDBC connector, at processing time for each event, optionally with a cache in front. It is simpler, but it puts load on that system and returns whatever the row says at the moment of the lookup.

Which version of the table?

A table changes over time, so a stream-table join depends on which version of each row it reads. Take an order in euros placed at 10:00, converted with an exchange rate that changes at 10:01.

Latest version (processing time) — Uses the 10:01 rate — Uses whatever the rate is next week
Version valid at the event's time (event time) — Uses the rate valid at 10:00 — Same answer as the first run

Reading the latest version is simple and often good enough for slowly changing data. Reading the version as of the event's timestamp gives the same answer every time you run it, which matters for billing, audits and backfills. Flink calls this an event-time temporal join against a versioned table (FOR SYSTEM_TIME AS OF orders.order_time); once an order is joined, later rate changes do not alter its result. Kafka Streams does the same when the table is backed by a versioned store: the join looks up the table as of the stream record's timestamp.

Table-table join: keep a view up to date

When both inputs are tables fed by changelogs, the join's output is a table too: a materialized view that changes whenever either side changes. A change to a customer's name updates every joined row for that customer; a new order adds one row.

The catch is state. A Flink regular join, the plain SQL JOIN over updating inputs, must keep both sides in state forever, because any old row might match a future change. Flink recommends setting a state time-to-live (TTL), accepting that a row dropped from state can no longer be matched. Kafka Streams offers KTable-KTable joins on the primary key and on a foreign key.

Event time, watermarks and late events

Stream joins that care about time need two clocks. Event time is when the event happened, written into the event by its source. Processing time is when the processor happens to handle it. A phone that was offline sends events minutes or hours late, so processing time gives answers that depend on the network.

A watermark is the processor's measure of progress in event time. Flink defines a watermark with time t as a statement that no more events with a timestamp at or before t should arrive. The watermark is what lets an interval join close a window and drop its state, and what triggers an event-time temporal join. An event that arrives after the watermark has passed its timestamp is late.

Waiting longer before advancing the watermark catches more late events but delays results and holds more state. There is no setting that is both complete and fast; you choose per use case.

What you will be able to do

Name the three kinds of stream join and the state each one keeps.
Design an enrichment pipeline that keeps a local copy of a table current through CDC.
Explain why a stream-table join should read the table version valid at the event's time when results must be repeatable.
Explain event time, watermarks and late events, and the trade-off in how long to wait.

Key concepts

Stream-stream join: Pairs events from two streams that share a key and fall within a time window. Unmatched events wait in state until the window closes.
Interval join: A stream-stream join whose condition bounds the time between the two events, so the processor can drop old events from state once the watermark passes.
Stream-table join: Enriches each event with the matching row of a table. The processor keeps a copy of the table, updated from its changelog. Only new events produce output.
Table-table join: Joins two changelogs into a materialized view that is updated when either table changes. Needs the current rows of both tables in state.
Temporal join: A stream-table join that reads the version of each table row valid at the event's timestamp, so reruns give the same result.
Event time: When the event happened, as recorded by its source. Processing time is when the processor handles it.
Watermark: A marker of progress in event time: a watermark at time t says no more events at or before t are expected. It closes windows and frees state.
Co-partitioning: Both join inputs split by the same key into the same number of partitions, so matching records reach the same processor instance.

When to use it

Designing a real-time pipeline that enriches events with customer, product or price data
Matching related events from two streams, such as impressions and clicks, or orders and payments
Keeping a denormalized view (a search index, a cache) in step with several source tables
Explaining why a streaming result differs when you rerun it, and how event time fixes that

Interview questions

Design a real-time pipeline that adds customer country and tier to every purchase event.
How would you compute click-through rate by joining ad impressions with clicks in real time?
Why use event time instead of processing time for stream joins?
A streaming SQL join has been running for a month and its state keeps growing. Why, and what do you do?