DE Mastery GitHub

Apache Flink · FLK-03

Time & windows

From your world. late-arriving facts and the as-of grain in your SCD2 / event marts

← All exercises

Hourly GMV per company with event time and a late side output

Compute GMV (sum of amounts) per company per event-time hour (here a short tumbling window). Capture late-arriving sales in a side output, so that Flink does not silently drop any of them. This is the streaming form of a late-arriving-fact mart with a reconciliation table.

Task

Implement de.flk.HourlyGmv.build(DataStream<Event> events, OutputTag<Event> lateTag, long windowMs).

The driver gives you a watermarked DataStream<Event> (each Event has companyId, amountCents, tsMs). Wire the windowing:

  1. keyBy the companyId.
  2. Apply a tumbling event-time window of size windowMs (TumblingEventTimeWindows.of(Duration.ofMillis(windowMs))).
  3. Route events that arrive after their window is closed to the late side output: sideOutputLateData(lateTag).
  4. Aggregate the amounts per window. Emit one row per (company, window) as companyId|windowStartMs|sumCents.

Return the windowed SingleOutputStreamOperator<String>. The driver reads its main output and its getSideOutput(lateTag) stream.

What the tests check

  • Per-(company, window) sums equal the sums of an independent oracle over the on-time events.
  • A late event (an event whose window already closed) appears in the late side output. It is not added to the window sum.
  • Windows use event time. A window start is always a multiple of windowMs, and the 1500ms event lands in [1000, 2000).

Hints

  • Flink 2.x uses java.time.Duration for window sizes. The old Time class is gone.
  • The late events show up on getSideOutput(lateTag). The driver reads them. You only need to call sideOutputLateData(lateTag) on the windowed stream.
  • Use a ProcessWindowFunction and ctx.window().getStart() for the window start.

Starter code

src/main/java/de/flk/HourlyGmv.java
package de.flk;

import de.flk.harness.Event;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * FLK-03 starter — NOT implemented yet (the tests must fail: G1).
 *
 * Window the event stream per company on a tumbling EVENT-TIME window of size
 * {@code windowMs}, aggregate the amount, and route late events to
 * {@code lateTag} via a late side output. Emit companyId|windowStartMs|sumCents.
 */
public class HourlyGmv {

    public static SingleOutputStreamOperator<String> build(
            DataStream<Event> events, OutputTag<Event> lateTag, long windowMs) {
        // TODO: keyBy(companyId) -> window(TumblingEventTimeWindows.of(..)) ->
        //       sideOutputLateData(lateTag) -> process(sum). For now, pass the
        //       raw stream through a trivial window-less map so the tests fail.
        return events.process(new org.apache.flink.streaming.api.functions.ProcessFunction<Event, String>() {
            @Override
            public void processElement(Event e, Context ctx, Collector<String> out) {
                out.collect(e.companyId + "|" + 0 + "|" + e.amountCents);
            }
        });
    }
}

Hints

Hint 1

keyBy the companyId, assign a TumblingEventTimeWindows.of(Duration.ofMillis(windowMs)), and aggregate the amount per window. Build windows with java.time.Duration (the old Time class is gone in Flink 2.x).

Hint 2

Call sideOutputLateData(lateTag) on the windowed stream so events arriving after the window closes go to the side output instead of being silently dropped.

Hint 3

Emit one result per (company, window) as companyId|windowStart|sumCents. Use a ProcessWindowFunction and ctx.window().getStart() for the window start.

Run it locally
uv run de start FLK-03/01-hourly-gmv
# edit workspace/FLK-03/01-hourly-gmv/
uv run de test