A join combines related records. In stream processing, it can enrich an incoming sale with product details, attach a customer to a loyalty card, or correlate a price change with a sale. Unlike a database query over stored tables, a streaming join must account for data that keeps arriving. section 12.4.3
Choose the join by the kind of data on each side:
- Stream–table: look up the stream record’s key in a table. A new stream record triggers the join; for example, sales data can acquire manufacturer details from a products table. section 12.4.3
- Table–table: join current values by key. Changes on either side trigger a lookup and update the result; customer and loyalty-card tables are one example. section 12.4.3
- Stream–stream: retain records from both streams for a limited time. Matching keys within the window can then be correlated; without a time limit, the other infinite stream cannot simply be looked up. section 12.4.3
Notify every active device
Key notifications and device records by user ID. Aggregate device updates into a table of each user’s active devices, then join incoming notifications with that table. The result carries the device list needed for delivery. The same flow can be expressed with Kafka Streams or with a Flink SQL view and join. section 12.4.4
KStream priceChanges = builder.stream(
"products.prices.changelog"); #1
KStream salesData = builder.stream("salesData"); #2
JoinWindows joinWindow = JoinWindows.ofTimeDifferenceWithNoGrace( #2
Duration.ofMinutes(1)); #2
ValueJoiner joiner = (priceChange, saleData) ->
new PriceSaleInfo(priceChange, saleData);
KStream correlated = priceChanges.join(salesData, joiner, joinWindow);
A Kafka Streams listing reads price changes and sales, sets a one-minute join window, and produces a stream of correlated records. section 12.4.3
Plan keys and partitioning before joining Kafka topics: related records must reach corresponding partitions. Use a stream-processing library rather than implementing a join in a basic consumer. Keep large, complex joins out of Kafka when possible: a relational database can perform them in SQL before the resulting data is sent to Kafka. section 9.4.2 section 18.2
