Latest valid location per driver
Python data engineering interview problem. Difficulty: intermediate. Pattern: Dictionaries. About 16 minutes. Part of the Pro drill bank.
Keep each driver's recent trail: drop repeated pings and pings more than 30 seconds older than that driver's newest ping. Treat this as a production helper: match the contracted return shape, including empty and duplicate inputs.
A location stream contains duplicate GPS pings and old positions. Dispatch only needs each driver's recent, distinct trail. Find each driver's newest timestamp, remove duplicate pings, keep pings no more than 30 seconds older than that newest one, and return sorted tuples grouped by driver. Write latest_locations(pings: list[dict]) -> dict. For d1 the newest time is 155, so 150 is kept, 100 is dropped, and the duplicate 150 is removed; d2 keeps its only ping. The starter file already contains the function name, harness and print lines. Keep them and fill only the function body.
Requirements
- Find the newest timestamp separately for each driver.
- Deduplicate identical driver, timestamp and coordinate records.
- Sort drivers and each driver's retained tuples.
Constraints
- A ping timestamp is an integer number of seconds.
- Exactly 30 seconds old is still valid and the return type is a dict.
Examples
Input: latest_locations([{"driver": "d1", "ts": 150, "lat": 1.2, "lon": 3.4}, {"driver": "d1", "ts": 100, "lat": 1.0, "lon": 3.0}, {"driver": "d1", "ts": 155, "lat": 1.3, "lon": 3.5}, {"driver": "d1", "ts": 150, "lat": 1.2, "lon": 3.4}, {"driver": "d2", "ts": 40, "lat": 9.0, "lon": 9.5}]) Output: {'d1': [(150, 1.2, 3.4), (155, 1.3, 3.5)], 'd2': [(40, 9.0, 9.5)]} d1's newest ping is 155, so its ping at 100 is too old and the repeated ping at 150 appears once. d2 has its own cutoff, so its only ping stays.
Topics: lakebench, python, streaming, dedupe, late data.
More Python interview questions · All interview problems · Learn data engineering
Latest valid location per driver
Interview-style drill: Keep each driver's recent trail: drop repeated pings and pings more than 30 seconds older than that driver's newest ping.
Implement `latest_locations(pings: list[dict]) -> dict`. A ping is `{"driver", "ts", "lat", "lon"}` with `ts` in whole seconds. Pings arrive in any order. For every driver, find the newest `ts` seen for that driver. Keep a ping only if it is not a repeat of an identical ping (same driver, ts, lat and lon) and it is at most 30 seconds older than that driver's newest ping (a ping exactly 30 seconds older is kept). Return `{driver: [(ts, lat, lon), ...]}` with each driver's pings sorted by `ts` and drivers sorted by name. The last tuple of a driver is their latest location.