Handling Duplicate, Late, and Out-of-Order Events

PYSPARK coding challenge · Difficulty: hard · Topic: Structured Streaming · +150 XP

Scenario

You are a Data Engineer at Uber processing real-time vehicle telemetry.

Vehicles stream events continuously. Networks retry, devices buffer, and radios

drop out — so events do not arrive in the order they happened, and the same

event can arrive more than once. Your pipeline has to report what each vehicle

actually did, minute by minute, regardless of when the rows turned up.

Two clocks matter, and confusing them is the whole problem:

`text

event_time -> when it happened -> decides which window the event belongs to

ingest_time -> when it arrived -> evidence of lateness only, never a grouping key

watermark -> how long we wait -> bounds state, and decides what is too late

`

Inputs

You are given two variables. There is deliberately no ready-made streaming

DataFrame — you build it yourself.

| Variable | What it is |
| --- | --- |
| `EVENTS_PATH` | Directory of JSON event files, written as three separate parts |
| `EVENT_SCHEMA` | The schema below |

| Column | Type | Description |
| --- | --- | --- |
| `event_id` | string | Globally unique event id |
| `vehicle_id` | string | Vehicle identifier |
| `event_time` | timestamp | When the event actually occurred |
| `speed` | double | Speed in km/h |
| `distance_km` | double | Distance covered |
| `ingest_time` | timestamp | When the event reached the streaming system |

Task

Build a Structured Streaming pipeline producing one row per vehicle per

1-minute event-time window:

| Column | Meaning |
| --- | --- |
| `vehicle_id` | Vehicle |
| `window_start` | Window start (event time) |
| `window_end` | Window end (event time) |
| `total_events` | Count of **unique** events |
| `total_distance_km` | Sum of `distance_km`, rounded to 2 dp |
| `avg_speed` | Mean `speed`, rounded to 2 dp |
| `max_speed` | Maximum `speed` |

Assign the result to df_result, ordered by vehicle_id, window_start, and

finish with df_result.show().

Requirements

1. Read with readStream using EVENT_SCHEMA, and set maxFilesPerTrigger to 1 so each file arrives as its own micro-batch. A watermark does nothing inside a single batch — it is derived from the previous batch — so without this the streaming semantics are untestable.

2. Apply a 10-minute watermark on event_time before deduplicating. The watermark is a *policy*, not an assertion: it tells Spark how long to keep deduplication keys and open windows in the state store, and therefore how late an event may be and still be counted. Ten minutes here means "we accept telemetry up to ten minutes late, and we are willing to hold ten minutes of state to do it". Set it too tight in production and genuinely late events are silently discarded; set it to nothing and the state store grows without bound.

3. Deduplicate on event_id only. The same event is retried with a *different* ingest_time, so deduplicating on the whole row keeps both copies.

4. Deduplicate before aggregating. A duplicate counted into a window cannot be removed afterwards.

5. Window on event_time, never ingest_time.

6. Write with outputMode("append") and trigger(availableNow=True) to a memory sink, then query that table. In complete mode Spark retains every group forever and never drops late rows, so the watermark has no observable effect at all.

7. Build the stream inside a function. This platform auto-displays module-level DataFrames, and doing that to a streaming DataFrame raises AnalysisException: Queries with streaming sources must be executed with writeStream.start(), which fails the run even when your logic is correct. Keep the streaming DataFrame and the query local to the function so they go out of scope.

The data

Three files, arriving as three micro-batches:

part-01 — five rows, deliberately *not* in event-time order: E105 at

10:01:30 is written before E101 at 10:00:05. E102 appears twice, at

ingest_time 10:00:21 and 10:00:25 — one event, retried.

part-02 — E103, which happened at 10:00:10 but did not arrive until

10:05:00. Five minutes late, and after events that happened later. It still

belongs in the 10:00–10:01 window.

part-03 — E106 from V103 at 10:15:00. Its job is to advance the

watermark past the earlier windows so append mode can emit them. Its own

window is still open when the stream ends, so V103 is absent from the

output — which is correct, not a bug.

Worked example

For V101 in the 10:00–10:01 window, after deduplication the surviving events

are E101, E102 and E103:

`text

total_events = 3 (not 4 -- E102 arrived twice)

total_distance_km = 0.8 + 1.1 + 0.9 = 2.8

avg_speed = (50.0 + 55.0 + 52.0)/3 = 52.33

max_speed = 55.0

`

Count 4 means you deduplicated on the whole row instead of event_id, or

aggregated before deduplicating. Losing E103 from this window means you

grouped on ingest_time, which would put it in a 10:05 bucket.

Expected output

`text

+----------+-------------------+-------------------+------------+-----------------+---------+---------+

|vehicle_id|       window_start|         window_end|total_events|total_distance_km|avg_speed|max_speed|

+----------+-------------------+-------------------+------------+-----------------+---------+---------+

|      V101|2026-03-01 10:00:00|2026-03-01 10:01:00|           3|              2.8|    52.33|     55.0|
|      V101|2026-03-01 10:01:00|2026-03-01 10:02:00|           1|              1.2|     60.0|     60.0|
|      V102|2026-03-01 10:01:00|2026-03-01 10:02:00|           1|              0.7|     45.0|     45.0|

+----------+-------------------+-------------------+------------+-----------------+---------+---------+

`

What this PYSPARK challenge teaches you

“Handling Duplicate, Late, and Out-of-Order Events” is a hard-level PYSPARK challenge focused on Structured Streaming. Working through it gives you hands-on practice with structured-streaming, watermark, deduplication, event-time, late-data, out-of-order — the kind of transformation you are asked to write in real data engineering work and in technical interviews. You can solve it directly in the browser: the dataset is pre-loaded, so you write the query or DataFrame code, run it, and compare your output against the expected result immediately.

Concepts covered

• structured-streaming
• watermark
• deduplication
• event-time
• late-data
• out-of-order
• windowing
• uber

How to approach it

If you get stuck, work through these steps in order before looking at a full solution — each one narrows the problem down:

1. Do not let a streaming DataFrame survive to the end of your script. Build it inside a function; the compiler auto-displays module-level DataFrames and that raises AnalysisException on a streaming one.
2. Set maxFilesPerTrigger to 1. With everything in one micro-batch the watermark is never consulted, because it is computed from the previous batch. Choose the watermark for the lateness you are willing to accept and the state you are willing to hold -- it bounds the dedup keys Spark must remember.
3. dropDuplicates(['event_id']), not dropDuplicates(). The retry of E102 differs in ingest_time, so the whole-row form treats the two copies as distinct events.
4. Group on F.window('event_time', '1 minute'). ingest_time only tells you when the row arrived, which is not what the report is about.
5. Use outputMode('append') with trigger(availableNow=True). complete mode keeps every group forever and never drops late data, so the watermark becomes decorative.

Where this comes up

Variations of this problem have been reported in interviews at Uber. Interviewers use it to check whether you can express the logic cleanly and reason about correctness on edge cases such as ties, nulls and empty groups.

How to practise it on PySpark.in

Open the challenge, write your PySpark code in the editor and press Run to execute it against the sample dataset. Submitting checks your output against every test case, including hidden ones, so you find out straight away whether your logic holds up. You can retry as often as you like, and each solved challenge adds to your XP.

Related PYSPARK challenges

• Fix the Broken Pipeline
• Optimize the 100M-Row Join
• Parse Apache Logs with Regex
• Optimize Small DataFrame Join with Broadcast
• Optimize Average Rating Calculation for Products
• Deduplicate and Aggregate User Actions with Latest Session

Helpful resources

• Open the notebook

Frequently asked questions

Do I need to install Spark or a database to solve this?

No. The PYSPARK environment runs in your browser with the sample data already loaded, so there is nothing to install or configure.

Is this challenge free?

Yes - the problem, the sample dataset, the hints and unlimited test runs are free.

What level is it?

It is rated hard and covers Structured Streaming.

Solve this challenge free on PySpark.in