Joins — regular, interval, temporal (point-in-time) & lookup

5 h
requires FLK-03

From your world. SCD2 point-in-time (as-of) joins: enrich a fact with the dimension version that was effective at the event's business time

← All exercises

Point-in-time (as-of) enrichment of orders with FX rate and address

Enrich each order with the USD-converted amount and the city of the customer as they were in effect at the event time of the order. This is the streaming form of an SCD2 as-of join. Converting at today's rate or with today's address is a SOX accuracy defect. Your job must pick the version that was effective then.

Implement de.flk.AsOfEnrich:

public static Map<Integer, String[]> run(
        List<String[]> orders,     // {orderId, custId, ccy, amtMinor, tsMillis}
        List<String[]> fxRates,    // {ccy,    rateMicros, effFromMillis}
        List<String[]> addresses)  // {custId, city,       effFromMillis}
    throws Exception

All fields arrive as strings (integers as decimal strings). Build a Flink DataStream job that does these steps:

Parse the three lists into streams. Each order carries event time tsMillis. Each FX version and each address version carries event time effFromMillis. Assign a monotonic watermark on each stream from that timestamp.
For every order, find:
- the FX rate version for the currency of the order with the greatest effFromMillis <= order ts.
- the address version for the customer of the order with the greatest effFromMillis <= order ts.
Compute usd = amtMinor * rateMicros. Keep it as a long of micro-dollar-minor-units (see below). Then the arithmetic is exact and the oracle can match it precisely.
Return a map orderId -> {usdString, city}. Here usdString is String.valueOf(usd). The value city is the as-of city. Use the literal string "NULL" if no address version is effective yet.

Exact arithmetic (so the oracle matches with no float tolerance)

amtMinor is the order amount in minor units (for example cents) as a long.
rateMicros is the FX rate × 1_000_000 as a long (so rate 1.25 is 1250000).
usd = amtMinor * rateMicros (a long). Do not divide. The test compares the raw product, so there is no rounding to disagree about.

Suggested approach (DataStream)

Key the orders stream and each dimension stream by the join key. connect the orders with the FX stream. Also connect them with the address stream (in a separate function or in the same function). Use a KeyedCoProcessFunction. Store dimension versions in ListState. Buffer each order and register an event-time timer at the order timestamp. In onTimer, pick the as-of version(s) and emit.

A Flink SQL / Table API temporal join (FOR SYSTEM_TIME AS OF) is a valid alternative design. But the as-of rule and the watermark dependency are the point of this exercise. Start with the DataStream version.

Starter code

src/main/java/de/flk/AsOfEnrich.java

package de.flk;

import java.util.*;

/**
 * Starter: not implemented. Returns an empty map so the tests fail (G1).
 *
 * Build a Flink DataStream job that, for each order, picks the FX rate version
 * and address version whose effective_from is the greatest value <= the order's
 * event time (the as-of rule), then returns orderId -> {usdString, city}.
 */
public class AsOfEnrich {
    public static Map<Integer, String[]> run(
            List<String[]> orders, List<String[]> fxRates, List<String[]> addresses)
            throws Exception {
        // TODO: key both sides by the join key, connect with a
        // KeyedCoProcessFunction, buffer versions in ListState, buffer each
        // order and emit on an event-time timer at the order timestamp.
        return new LinkedHashMap<>();
    }
}

Hints

Hint 1

Key BOTH the orders stream and each dimension stream by their join key (currency for FX, customer id for address), then connect them with a KeyedCoProcessFunction so related rows share keyed state.

Hint 2

Buffer dimension versions as (value, effective_from) in ListState. For each order, buffer it and register an event-time timer at the order's timestamp; emit only when the timer fires (watermark reached it).

Hint 3

In onTimer, the as-of version is the one with the GREATEST effective_from that is <= the order timestamp. A strict '<' or picking the newest version is the classic bug.

Run it locally

uv run de start FLK-05/01-asof-enrichment
# edit workspace/FLK-05/01-asof-enrichment/
uv run de test