RouteMatching

Snapping Irregular Fixes to a Common Time Grid with Polars

The pandas resampler in fixed-interval resampling is the right tool for one vehicle-day. It stops being the right tool somewhere around ten million fixes, at which point the per-group Python call overhead dominates and the whole frame has to be resident. This page rewrites the same operation as a single lazy Polars plan that expresses the per-vehicle grid, the alignment and the gap classification as one query, and streams it.

The technique is the same as in the parent trace resampling topic: align real fixes onto slots, fill only short holes, label everything. What changes is that none of it runs in a Python loop.

Compatibility and Configuration Requirements

Requirement Value Notes
Polars ≥ 1.0 join_asof gained by with tolerance semantics that are stable from 1.0
Input format Parquet, partitioned by operating date Lets the scan skip whole days
Timestamp dtype Datetime("us", "UTC") Mixing tz-naive and tz-aware columns raises at plan time, which is the good outcome
Sort state Sorted by (vehicle_id, ts_utc) Required by join_asof; enforce it, do not assume it
Memory ~2 GB for a 40 M-fix month in streaming mode Without streaming, roughly 9 GB

A Self-Contained Fleet Aligner

from __future__ import annotations

import polars as pl

GRID = "30s"
TOL = "10s"
GAP_LIMIT = 2


def build_grids(fixes: pl.LazyFrame, every: str = GRID) -> pl.LazyFrame:
    """One grid per (vehicle_id, operating_date), bounded by that day's own fixes."""
    return (
        fixes.group_by(["vehicle_id", "operating_date"])
        .agg(
            pl.col("ts_utc").min().dt.truncate(every).alias("t0"),
            pl.col("ts_utc").max().dt.truncate(every).alias("t1"),
        )
        .with_columns(
            pl.datetime_ranges("t0", "t1", interval=every, time_zone="UTC").alias("slot_utc")
        )
        .explode("slot_utc")
        .drop(["t0", "t1"])
    )


def align_fleet(
    fixes: pl.LazyFrame,
    every: str = GRID,
    tolerance: str = TOL,
    gap_limit: int = GAP_LIMIT,
) -> pl.LazyFrame:
    """Snap every vehicle-day onto its own grid and classify each slot."""
    grids = build_grids(fixes, every).sort(["vehicle_id", "slot_utc"])
    src = fixes.sort(["vehicle_id", "ts_utc"])

    aligned = grids.join_asof(
        src,
        left_on="slot_utc",
        right_on="ts_utc",
        by="vehicle_id",
        strategy="nearest",
        tolerance=tolerance,
    ).with_columns(pl.col("lat").is_not_null().alias("observed"))

    # A null streak shares a value of `_run`, so its length is a window count.
    classified = (
        aligned.with_columns(
            pl.col("observed").cum_sum().over(["vehicle_id", "operating_date"]).alias("_run")
        )
        .with_columns(
            pl.len().over(["vehicle_id", "operating_date", "_run"]).alias("_run_len")
        )
        .with_columns(
            pl.when(pl.col("observed"))
            .then(pl.lit("observed"))
            .when(pl.col("_run_len") <= gap_limit + 1)
            .then(pl.lit("interpolated"))
            .otherwise(pl.lit("missing"))
            .alias("provenance")
        )
    )

    return (
        classified.with_columns(
            pl.when(pl.col("provenance") == "missing")
            .then(None)
            .otherwise(pl.col("lat").interpolate())
            .over(["vehicle_id", "operating_date"])
            .alias("lat"),
            pl.when(pl.col("provenance") == "missing")
            .then(None)
            .otherwise(pl.col("lon").interpolate())
            .over(["vehicle_id", "operating_date"])
            .alias("lon"),
        )
        .drop(["_run", "_run_len", "ts_utc"])
    )

Run it against a partitioned dataset:

out = (
    align_fleet(pl.scan_parquet("s3://fleet/fixes/date=*/*.parquet"))
    .collect(engine="streaming")
)
out.write_parquet("resampled/", partition_by=["operating_date"])
Per-vehicle grids against one fleet-wide grid Three vehicles active for different parts of a day. A single fleet-wide grid emits a slot for every vehicle across the whole 24 hours, most of which are null. Per-vehicle grids emit slots only between each vehicle's first and last fix, which is roughly a fifth of the rows for a typical delivery fleet. Rows emitted for three vehicles over one 24-hour day fleet-wide grid v-014 v-015 v-027 per-vehicle grids Faint bars are emitted rows with no observation anywhere near them — 78 % of the fleet-wide output for this day.

Execution and Tuning Guidelines

Partition the scan, not just the output. scan_parquet over a date-partitioned layout lets the predicate push down so a backfill of one week reads one week. Without partitioning, every run reads the whole archive and the streaming engine spends its time on I/O it did not need.

Set tolerance in the same units as the grid. Polars accepts a duration string, and a mismatch — every="30s" with tolerance="10m" — will not error. It will quietly move fixes by up to ten minutes, which at motorway speed is fourteen kilometres.

by="vehicle_id" is not optional. Without it, join_asof will happily match a slot for one vehicle against the nearest fix belonging to a different vehicle. The result is a table that looks complete and is wrong in a way no null check will find.

Watch the explode. datetime_ranges followed by explode materialises every slot. For a fleet whose vehicles are active twelve hours a day on a 30-second grid, that is 1,440 rows per vehicle-day — manageable. On a 1-second grid it is 43,200, and the explode becomes the memory peak of the whole plan. If you need a 1 Hz grid, partition the work by vehicle-day before the explode rather than after.

Eager against streaming collection Peak resident memory for the same alignment plan at one, ten and forty million input fixes. Eager collection scales roughly linearly and reaches nine gigabytes at forty million; streaming stays under two gigabytes across the range, at a cost of about fifteen percent in wall-clock time. Peak resident memory, same plan, two collection modes 10 GB 5 GB 0 1 M fixes 10 M fixes 40 M fixes eager streaming Streaming costs roughly 15 % in wall clock and removes the memory cliff that makes a monthly backfill fail at 03:00. Below about five million fixes the eager path is simpler and the difference does not matter — use it.

Common Pitfalls Specific to This Technique

Assuming the input is sorted because it came from Parquet. Partition order is not row order. join_asof raises InvalidOperationError: argument in operation 'asof_join' is not sorted, which is the good case; the bad case is a file that happens to be sorted today and is not after a compaction. Sort explicitly.

Interpolating across the vehicle-day boundary. pl.col("lat").interpolate() without an over clause will happily draw a line from one vehicle’s last fix of Tuesday to another vehicle’s first fix of Wednesday. The over(["vehicle_id", "operating_date"]) in the implementation above is what prevents it, and removing it produces output that looks fine until someone plots it.

Letting operating_date come from UTC. If the column is derived as ts_utc.dt.date(), a fleet operating across two time zones will have its night shifts split at UTC midnight, and the per-vehicle grid will emit two half-days with a seam in the middle. Derive the operating day in the vehicle’s own zone, as covered in calculating accurate dwell times across timezone shifts.

Reading the Query Plan Before Running It

Polars will tell you what it intends to do, and for a plan with a join_asof, several window expressions and an explode, that is worth reading before committing a fleet-month to it.

plan = align_fleet(pl.scan_parquet("s3://fleet/fixes/date=*/*.parquet"))
print(plan.explain(optimized=True))

Three things are worth looking for.

Whether the projection pushed down. If the plan still reads twenty columns when the function uses six, add an explicit select early. On a wide telematics table — and they are always wide, because every CAN signal ends up in it — that alone can halve the I/O.

Whether the date predicate pushed into the scan. A backfill restricted to one week should show a partition filter in the scan node. If it does not, the filter is being applied after the read and the job is scanning the whole archive to throw most of it away.

Where the explode sits relative to the join. The exploded grid is the widest intermediate in the plan. If the optimiser has moved a filter after it, the job materialises rows it is about to discard. Reordering the source expression so the filter precedes the range generation usually fixes it.

The streaming engine does not support every operation, and an unsupported node silently falls back to in-memory execution for that part of the plan — which is exactly the behaviour that turns a job that worked in testing into one that is killed by the scheduler at 03:00. Check for the fallback explicitly rather than discovering it from a memory graph:

explained = plan.explain(optimized=True, streaming=True)
if "STREAMING" not in explained:
    raise RuntimeError("plan is not streamable — inspect before scheduling a backfill")

Size the job from the widest intermediate, not the input. A month of 30-second fixes for eight hundred vehicles is roughly forty million rows on the way in and rather more than that after the grid is exploded, because grids cover the whole active period including the gaps the fixes do not. Working memory should be budgeted against the second number.

Where the rows are in the plan Row counts through the plan for one fleet-month: forty million input fixes, forty-four million grid slots after the explode, forty-four million after the join, and thirty-nine million surviving output rows once missing slots are dropped by the consumer. The exploded grid is the widest point. Rows at each stage, one fleet-month scan_parquet 40 M datetime_ranges 24 k groups explode 44 M — widest join_asof 44 M non-missing output 39 M Budget memory against the widest bar, not the first one — the grid is larger than the data it aligns.

Related

Up to section Trace Resampling and Densification for Fleet GPS Related Resampling GPS Traces to a Fixed Interval in pandas Related Choosing an Interpolation Limit for Fleet GPS Gaps Related Douglas-Peucker Simplification for Fleet Trace Storage Related Deduplicating Repeated GPS Fixes from Buffered Trackers