The Stream Joiner

A hard Python interview practice problem on DataDriven. Write the Python and run it against test cases, with instant feedback.

Domain
Python
Difficulty
hard
Seniority
L5
Asked in
Google

Problem

You're correlating two time-sorted event streams where each event is a dict carrying a timestamp and a value, given a numeric tolerance. Walking stream_a in order, pair each event with the single nearest event in stream_b whose timestamp lies within tolerance of it, emitting a dict of a_value, b_value, and the absolute gap; an A event with nothing inside its window contributes no row. When two B events are equally near, keep the earlier one, and remember a single B event can be the nearest match for several A events.

Events don't wait for each other. This does.

Example

Input:

stream_a = [{"value":"a1","timestamp":1},{"value":"a2","timestamp":5},{"value":"a3","timestamp":12}], stream_b = [{"value":"b1","timestamp":2},{"value":"b2","timestamp":4.5},{"value":"b3","timestamp":20}], tolerance = 3

Output:

[{"a_value":"a1","b_value":"b1","gap":1},{"a_value":"a2","b_value":"b2","gap":0.5}]

Worked solution and explanation

What this really is

Strip away the streaming language and this is a nearest-neighbor search between two sorted sequences. Anyone can pair every A with every B and keep the closest, and it even passes the tests. The thing that separates candidates is exploiting the sort: you never need to rescan B from the start. A shared pointer that only crawls forward turns an O(a*b) double loop into a single linear pass. Get that pointer wrong and you either rescan everything, or worse, advance it past B events that a later A event still needs.

Break down the requirements

Step 1: Iterate through stream A events

For each event in A, find the closest match in B within the allowed time difference and emit a dict with a_value, b_value, and gap.

Step 2: Use a pointer into B to skip events that are too early

Since both streams are sorted, once a B event is earlier than a_time minus tolerance it is too early for this A and for every later A, so the left pointer can advance past it permanently.

Step 3: Track the closest match within the tolerance

Among all B events inside the window [a_time - tolerance, a_time + tolerance], pick the one with the smallest absolute time difference; ties go to the earliest, which the forward scan hits first. Break out once a B event is past a_time + tolerance.

Step 4: Emit the joined dicts

If a match was found, append the dict with gap as a float; otherwise skip that A event. B events are never consumed, so one B can match several A events.

The solution

Two-pointer scan with tolerance window
def windowed_join(stream_a: list, stream_b: list, tolerance: float) -> list:
    matches = []
    left = 0
    for a_event in stream_a:
        a_time = a_event["timestamp"]
        # Advance the shared pointer past B events too early for this (and every later) A.
        while left < len(stream_b) and stream_b[left]["timestamp"] < a_time - tolerance:
            left += 1
        best_b = None
        best_gap = float("inf")
        j = left
        while j < len(stream_b):
            b_time = stream_b[j]["timestamp"]
            if b_time > a_time + tolerance:
                break
            gap = abs(a_time - b_time)
            if gap < best_gap:
                best_gap = gap
                best_b = stream_b[j]
            j += 1
        if best_b is not None:
            matches.append({
                "a_value": a_event["value"],
                "b_value": best_b["value"],
                "gap": float(best_gap),
            })
    return matches

Common follow-up questions