Incremental Toll Processing with Late-Arriving Events (PySpark)

PYSPARK coding challenge · Difficulty: hard · Topic: Incremental Processing · +150 XP

Scenario

You are a Data Engineer at Uber processing toll-plaza crossings.

The toll_events table grows continuously and holds billions of rows. Re-reading

all of history on every run costs hours of compute, so the pipeline runs

incrementally: each run processes only what changed since the last successful run.

Two things make that harder than it looks.

Events arrive late. A toll plaza can be offline for hours. An event that *happened*

on January 9 may not *arrive* until January 10. Its toll_time says January 9; its

event_updated_at says January 10.

Events get restated. A plaza can resend a crossing with a corrected distance. The

same event_id then appears twice with different event_updated_at values, and only

the newest one is true.

Table: toll_events

| Column | Type | Description |
| --- | --- | --- |
| `event_id` | long | Toll event ID. **Not unique in the table** — a restated event reappears with the same id |
| `vehicle_id` | string | Vehicle identifier |
| `toll_time` | timestamp | When the vehicle actually crossed the toll |
| `toll_plaza_id` | string | Toll plaza identifier |
| `distance_km` | double | Distance since the previous toll |
| `event_updated_at` | timestamp | When the row was created **or last corrected** |

You are also given the watermark from the last successful run:

`python

LAST_PROCESSED_AT = '2024-01-10 00:00:00'

`

Task

Using the PySpark DataFrame API, produce the daily vehicle statistics that this

run must rewrite:

| Column | Meaning |
| --- | --- |
| `vehicle_id` | Vehicle |
| `event_date` | Calendar date from `toll_time` |
| `total_events` | Number of toll crossings in that group |
| `total_distance_km` | Sum of `distance_km`, rounded to 2 decimals |
| `avg_distance_km` | Mean `distance_km`, rounded to 2 decimals |

Order by vehicle_id, then event_date.

Rules

1. Only changed data drives the run. An event participates in change detection only

when event_updated_at >= LAST_PROCESSED_AT.

2. event_date comes from toll_time, never from event_updated_at.

3. Deduplicate on event_id, keeping the row with the greatest event_updated_at.

4. Emit only affected groups. A (vehicle_id, event_date) group appears in the

output only if at least one changed event falls in it. Untouched days must not

appear — rewriting them is the full-scan cost you are avoiding.

5. Each emitted group must be complete. Its totals cover *every* event in that group

from all of history, not only the changed ones. A partial total silently corrupts

the table it overwrites.

6. Idempotent. Running the same batch twice must produce the same output.

7. DataFrame API only. No collect(), no toPandas(), no Python UDFs.

Worked example

LAST_PROCESSED_AT = '2024-01-10 00:00:00'. Four changed events cross the watermark:

| event_id | why it changed | group it touches |
| --- | --- | --- |
| 105, 106 | new crossings | `(V101, 2024-01-10)` |
| 107 | new crossing | `(V102, 2024-01-10)` |
| 108 | **late** — crossed Jan 9 23:55, arrived Jan 10 11:00 | `(V101, 2024-01-09)` |
| 102 | **restated** — distance corrected 20.0 to 25.0 | `(V101, 2024-01-09)` |

So three groups are affected. (V101, 2024-01-09) must be recomputed from all three

of its events — 101 (10.0), the corrected 102 (25.0), and the late 108 (5.0) — giving

3 events and 40.0 km. Aggregating only the two changed events would emit 2 and

30.0, and overwrite a correct row with a wrong one.

(V102, 2024-01-09) and (V103, 2024-01-08) had nothing change and must not appear.

(V104, 2024-01-07) is the boundary case: its only event was updated at exactly 2024-01-10 00:00:00. The rule is >=, so that day is in scope and V104 appears. Using > instead drops it silently.

Expected output

`text

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

|vehicle_id|event_date|total_events|total_distance_km|avg_distance_km|

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

|      V101|2024-01-09|           3|             40.0|          13.33|
|      V101|2024-01-10|           2|             20.5|          10.25|
|      V102|2024-01-10|           1|             15.0|           15.0|
|      V104|2024-01-07|           1|             18.0|           18.0|

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

`

What this PYSPARK challenge teaches you

“Incremental Toll Processing with Late-Arriving Events (PySpark)” is a hard-level PYSPARK challenge focused on Incremental Processing. Working through it gives you hands-on practice with incremental, late-arriving-data, idempotency, deduplication, watermark, cdc — 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

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. Two different timestamps do two different jobs. event_updated_at answers 'did this change?'; toll_time answers 'which day does it belong to?'. Swapping them puts the late event on the wrong date.
  2. Deduplicate before anything else. row_number() over a window partitioned by event_id and ordered by event_updated_at descending, keeping rank 1, gives you one row per event at its newest version.
  3. Change detection produces a set of GROUPS, not a set of events. Filter the deduped rows by the watermark, then select vehicle_id and event_date distinct -- that is the list of days this run must rewrite.
  4. Recompute each affected group from all of its history. A left_semi join from the deduped table back to the affected-group list keeps every row of an affected day and drops untouched days. Aggregating only the changed rows emits a partial total that overwrites a correct one.
  5. Idempotency falls out of the design: the output is a pure function of the table and the watermark, so the same batch always yields the same rows.

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

Helpful resources

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 Incremental Processing.

Solve this challenge free on PySpark.in