How Do You Test a Streaming Job?

A streaming job has 3 inputs that a batch job lacks: the order events arrive in, the gaps between them, and the clock the job consults to decide a window is complete. A test that fixes any of those to the laptop's wall clock passes locally and fails on the first evening the traffic pattern changes. Replay fixes all 3 at once: record a set of real events with their event time and their arrival time, feed them through the job with the clock driven by the fixture, and compare the output with a golden result a human checked by hand once.

3 layers of tests

• Unit tests on the pure functions: Parsing and enrichment each take an event and return a value, and so do key extraction and window assignment. Test all 4 with no engine running.
• Replay tests on the whole job: A recorded fixture of 20 to 200 events with event time and arrival time runs through the real job logic with a controlled watermark, and the output is compared to a golden file.
• A shadow run against the live log: The new version reads production events into a scratch sink while the old version keeps serving. A bounded recompute over the same window gives the expected totals.

A fixture that covers the cases that bite

A good fixture is small and adversarial. The one below uses 60 second windows and a watermark of the largest event time seen minus a 30 second lateness allowance. A window closes when the watermark passes its end, so window 0 stays open until an event at 90 or later appears, however late its own events arrive. Each row exercises 1 behavior: the ordinary case, a duplicate, an out-of-order event in an open window, and an event whose window has closed.

from collections import defaultdict


def run(events, window_s=60, lateness_s=30):
    seen, open_windows, closed = set(), defaultdict(float), {}
    watermark = 0
    for e in events:
        if e['id'] in seen:
            continue
        seen.add(e['id'])
        watermark = max(watermark, e['time'] - lateness_s)
        start = e['time'] - e['time'] % window_s
        if start in closed:
            continue
        open_windows[start] += e['amount']
        for w in list(open_windows):
            if w + window_s <= watermark:
                closed[w] = open_windows.pop(w)
    return closed


def test_late_and_duplicate():
    events = [
        {'id': 'a', 'time': 10, 'arrival': 11, 'amount': 5.0},
        {'id': 'b', 'time': 55, 'arrival': 56, 'amount': 7.0},
        {'id': 'b', 'time': 55, 'arrival': 58, 'amount': 7.0},
        {'id': 'c', 'time': 70, 'arrival': 72, 'amount': 2.0},
        {'id': 'd', 'time': 40, 'arrival': 85, 'amount': 3.0},
        {'id': 'e', 'time': 130, 'arrival': 131, 'amount': 1.0},
        {'id': 'g', 'time': 20, 'arrival': 140, 'amount': 9.0},
        {'id': 'f', 'time': 200, 'arrival': 201, 'amount': 4.0},
    ]
    assert run(events) == {0: 15.0, 60: 2.0}

The watermark only moves forward. Event d arrives at 85 with event time 40; the largest event time so far is 70, from c, so the watermark stands at 40, window 0 is still open, and its total becomes 15. Event g arrives at 140 with event time 20. By then e, with event time 130, has pushed the watermark to 100, window 0 has closed at 15, and g is dropped. Change lateness_s to 120 and g counts, window 0 closes at 24 once f arrives, and window 60 stays open because the watermark has only reached 80. The test tells you which business decision moved.

The shadow run, and the query that scores it

Replay proves the logic. A shadow run proves the job on production volume and partitioning, including the events that arrive late. Deploy the new version reading the live log into shadow_sink, leave it for 24 hours, then recompute the same 24 hours with a bounded run of the same code over the log into batch_recompute. The 2 should agree for every closed window. The query lists the disagreements, largest first, so the first 20 rows show whether the problem is a dropped partition or a late-data rule.

SELECT
  b.product_id,
  b.hour_start,
  b.order_count AS batch_count,
  coalesce(s.order_count, 0) AS stream_count,
  b.order_count - coalesce(s.order_count, 0) AS diff
FROM batch_recompute b
LEFT JOIN shadow_sink s
  ON b.product_id = s.product_id
  AND b.hour_start = s.hour_start
WHERE b.order_count <> coalesce(s.order_count, 0)
ORDER BY abs(b.order_count - coalesce(s.order_count, 0)) DESC
LIMIT 20

Reading the output follows a pattern. A diff on every key for 1 hour points at an offset or partition the shadow job skipped. A diff of a few units on random keys points at late events the shadow dropped and the bounded run kept, which means the lateness allowance is too short for real arrival delays. Zero rows for 24 hours is the release criterion.

Do

• Derive the watermark from the fixture's event times, never the machine clock
• Keep 1 golden file per scenario and name the scenario in the file
• Include a duplicate and a too-late event in every fixture
• Compare the shadow run to a bounded run of the same code

Avoid

• Test windows with sleep() calls and hope the timing holds
• Reuse the batch job's golden output for the stream without the lateness rule
• Declare a release good on throughput alone
• Let the fixture grow past a few hundred events

About this lesson

This section is part of the Batch or Stream?: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.