Streaming Map-Matching Pipelines

Batch map matching is comfortable: you read a completed trip, hand the whole observation sequence to a Viterbi decoder, and backtrack the single most probable path across the road network. Live fleet telematics offers no such luxury. Positions arrive continuously, one vehicle at a time, and the operations dashboard needs the matched road segment within a second or two — long before the trip is finished. You cannot wait for the end of a sequence that has no end.

Running map matching online over an unbounded GPS stream forces three changes on the classical hidden Markov model matcher: the decoder must work over a bounded sliding window instead of a whole trip, it must emit committed matches with a deliberate lag so that a later fix cannot silently rewrite history, and it must keep per-vehicle state alive across restarts, rebalances, and late-arriving packets. This page builds that pipeline on Kafka, partitioning by vehicle, keeping a windowed trellis per key, and emitting matched edges to a downstream topic with exactly the latency budget you configure. The two companion pages implement the windowed consumer itself and the exactly-once emission guarantees that keep each matched segment from being counted twice.

Why Naive Streaming Fails

The obvious approach — match each incoming fix independently to its nearest road edge — collapses at the first ambiguity. A single GPS observation between two parallel carriageways, or near a motorway on-ramp that shadows the through lane, has no way to choose the correct edge without the surrounding context. Snapping each point in isolation produces a matched trace that flickers between adjacent roads, and every flicker corrupts downstream mileage, toll, and speed-profile aggregation. The HMM formulation exists precisely to resolve that ambiguity by scoring whole paths, not points.

The second naive approach — buffer the whole trip and run batch Viterbi when it ends — restores accuracy but destroys latency. A vehicle on a four-hour long-haul route would produce no matched output until the engine cut off, which is useless for live geofencing, ETA, or exception alerting. Worse, “trip end” is itself a detection problem on a stream: you only know a trip finished once enough idle time has elapsed.

Streaming map matching threads between these failures. It keeps enough recent context to decode correctly, but bounds that context so per-fix cost stays constant, and it publishes results continuously while accepting that the newest fixes near the window head are provisional until the decoder has seen a few more observations.

Prerequisites

Message broker. Apache Kafka 2.8+ (or a compatible broker). The source topic must be keyed by vehicle_id so all fixes for one vehicle route to the same partition; this is what makes per-vehicle state local and lock-free. Plan partition count for peak concurrent vehicles, not average — a partition is the unit of parallelism and of state ownership.

Python client. confluent-kafka ≥ 2.3 (librdkafka-backed, required later for transactional exactly-once output) or kafka-python ≥ 2.0 for simpler at-least-once deployments. Python 3.10+, numpy ≥ 1.24 for the log-space trellis arithmetic.

A windowing strategy. Decide up front whether your window is count-based (last N fixes) or time-based (last T seconds of event time). Count windows are simpler and bound memory directly; time windows are more robust to variable sampling rates across a mixed fleet.

Per-vehicle keyed state. A place to hold each vehicle’s window buffer and Viterbi trellis. In-process dictionaries keyed by vehicle_id are fine for a single consumer, but they must be backed by a Kafka changelog topic or an external store (RocksDB, Redis) so that a rebalance does not lose the trellis.

A routing engine. An OSRM or Valhalla instance for candidate-edge lookups. Clean, deduplicated fixes help: apply Kalman filtering for GPS noise reduction upstream so the emission model is not fighting multipath spikes.

The Streaming Workflow

Step 1 — Partition the source topic by vehicle

Keyed partitioning is the load-bearing decision of the whole design. When the producer sets the Kafka message key to vehicle_id, the default partitioner hashes it so that every fix for a given vehicle is totally ordered on one partition and owned by exactly one consumer instance. That gives you single-writer access to that vehicle’s state without distributed locks.

from confluent_kafka import Producer

producer = Producer({"bootstrap.servers": "broker:9092", "linger.ms": 20})

def publish_fix(fix: dict) -> None:
    # key = vehicle_id ensures per-vehicle ordering and state locality
    producer.produce(
        topic="gps.fixes",
        key=fix["vehicle_id"].encode("utf-8"),
        value=encode(fix),  # value carries lon, lat, event_ts, hdop
    )
    producer.poll(0)

Expected shape: each partition carries a per-vehicle-ordered sub-stream of fixes; a consumer assigned partition p owns the state for every vehicle whose key hashes to p.

Step 2 — Maintain a sliding window of recent fixes per key

For each vehicle_id, keep a bounded deque of the most recent fixes. On every arrival, append the new fix, drop anything older than the window horizon, and re-sort by event time to absorb small out-of-order jitter within the buffer.

from collections import deque, defaultdict

class VehicleWindow:
    """Bounded per-vehicle context for windowed matching."""

    def __init__(self, max_fixes: int = 40, max_span_s: float = 120.0):
        self.max_fixes = max_fixes          # count bound on the window
        self.max_span_s = max_span_s        # time bound on the window
        self.fixes: deque = deque()

    def add(self, fix: dict) -> None:
        self.fixes.append(fix)
        # keep event-time order so the trellis columns are monotonic
        self.fixes = deque(sorted(self.fixes, key=lambda f: f["event_ts"]))
        # evict by count
        while len(self.fixes) > self.max_fixes:
            self.fixes.popleft()
        # evict by time span
        newest = self.fixes[-1]["event_ts"]
        while self.fixes and newest - self.fixes[0]["event_ts"] > self.max_span_s:
            self.fixes.popleft()

windows: dict[str, VehicleWindow] = defaultdict(VehicleWindow)

Expected shape: windows[vehicle_id].fixes is an event-time-sorted deque of at most max_fixes observations spanning at most max_span_s seconds.

Step 3 — Run windowed HMM matching

Over the window, build the standard HMM: emission probability from each fix to nearby candidate road segments (Gaussian in the great-circle distance to the snapped point) and transition probability between consecutive candidates (comparing the on-road routed distance to the straight-line displacement). Decode in log-space with a forward pass; the mechanics are identical to the batch HMM matcher, only the sequence length is bounded.

import numpy as np

def forward_logprob(window, candidates, sigma_z=6.0, beta=0.9):
    """
    One forward pass of Viterbi in log-space over the window.

    candidates[i] -> list of candidate edges for fix i
    sigma_z : emission std dev in metres (GPS accuracy)
    beta    : transition decay; larger tolerates bigger route/GPS gaps
    Returns the trellis of best log-scores and back-pointers.
    """
    n = len(window.fixes)
    trellis = [dict() for _ in range(n)]
    backptr = [dict() for _ in range(n)]

    for j, cand in enumerate(candidates[0]):
        emit = -0.5 * (cand.gps_dist_m / sigma_z) ** 2
        trellis[0][cand.edge_id] = emit
        backptr[0][cand.edge_id] = None

    for i in range(1, n):
        for cand in candidates[i]:
            emit = -0.5 * (cand.gps_dist_m / sigma_z) ** 2
            best_score, best_prev = -np.inf, None
            for prev in candidates[i - 1]:
                # transition: |routed_dist - haversine_dist| penalised exponentially
                d = abs(cand.route_dist_m[prev.edge_id] - cand.gc_dist_m[prev.edge_id])
                trans = -d / beta
                s = trellis[i - 1][prev.edge_id] + trans + emit
                if s > best_score:
                    best_score, best_prev = s, prev.edge_id
            trellis[i][cand.edge_id] = best_score
            backptr[i][cand.edge_id] = best_prev
    return trellis, backptr

Expected shape: a trellis with one column per fix in the window and one log-score per candidate edge, plus back-pointers for path recovery.

Step 4 — Emit matched edges with a deliberate lag

Here is the crux of online decoding. Backtrack from the best terminal state, but do not emit the whole path. The tail of the path — the fixes nearest the window head — is still volatile; a single new observation can flip a candidate. Only emit edges for fixes older than a lag horizon L, where the decoded state has converged and future observations will not revise it. This is lagged Viterbi backtracking, sometimes called fixed-lag smoothing.

def committed_edges(trellis, backptr, window, lag: int = 8, emitted_upto=None):
    """
    Backtrack and yield only edges older than the lag horizon.

    lag : number of most-recent fixes held back as provisional.
    emitted_upto : event_ts of the last already-emitted fix (idempotency guard).
    """
    n = len(window.fixes)
    if n <= lag:
        return []  # not enough context yet to commit anything

    # best terminal state
    last = max(trellis[-1], key=trellis[-1].get)
    path = [None] * n
    edge = last
    for i in range(n - 1, -1, -1):
        path[i] = edge
        edge = backptr[i][edge]

    out = []
    for i in range(n - lag):  # only fixes at or before the lag horizon
        fix = window.fixes[i]
        if emitted_upto is not None and fix["event_ts"] <= emitted_upto:
            continue  # already produced; do not double-count
        out.append({"vehicle_id": fix["vehicle_id"],
                    "event_ts": fix["event_ts"],
                    "edge_id": path[i]})
    return out

Expected shape: a list of newly committed (vehicle_id, event_ts, edge_id) records, each emitted exactly once as the window slides forward.

Step 5 — Handle late and out-of-order events with watermarks

Cellular buffering, tunnel reconnects, and retransmits mean fixes do not arrive in event-time order. Track a per-key watermark: the highest event time seen minus an allowed-lateness slack. A fix whose timestamp is behind the watermark is late. If it still falls inside the retained window, re-insert it and recompute the affected trellis columns; if it is older than the window, side-output it to a late.fixes topic for audit rather than corrupting an already-committed match.

def classify_event(window, fix, allowed_lateness_s=10.0):
    if not window.fixes:
        return "on_time"
    watermark = window.fixes[-1]["event_ts"] - allowed_lateness_s
    if fix["event_ts"] >= watermark:
        return "on_time"        # within lateness slack: reprocess window
    oldest = window.fixes[0]["event_ts"]
    return "reprocessable" if fix["event_ts"] >= oldest else "too_late"

Step 6 — Checkpoint state, then commit offsets

Order matters for correctness. Produce the matched edges, persist the updated window and trellis to the changelog, and only then commit the source offset. If the process dies between emit and commit, the fixes are re-read and re-matched; the emitted_upto guard in Step 4 (and, more strongly, the transactional producer) makes that replay harmless. Committing offsets before emitting is the classic way to lose matches on a crash — never do it.

Latency, Window Length, and Accuracy

Three parameters trade off against each other, and tuning them is the core operational skill for a streaming matcher.

• Window length W (fixes or seconds). Longer windows give the decoder more context and more stable paths, at higher per-fix CPU cost and memory. Below a floor, ambiguous parallel-road segments are decoded from too little evidence and flicker.
• Lag horizon L (fixes held back). This is your accuracy/latency dial. Emitted edges older than L are effectively final; the probability that a fix at lag L is later revised falls roughly geometrically in L once past the local ambiguity length. Emission latency is approximately L / sampling_rate plus processing time.
• Emission latency budget. For 1 Hz data, L = 8 yields around 8 s of matched-edge lag — acceptable for geofencing and ETA, marginal for instantaneous alerting. Drop L and accept a higher revision rate, or raise the sampling rate.

A useful rule: set W to at least twice the longest ambiguity you expect (the length of the longest stretch of indistinguishable parallel roads, in fixes), and set L to the ambiguity length itself. Measure the revision rate — the fraction of emitted edges that a later fix would have changed — in shadow mode before committing to a value. Because the forward pass is log-space, there is no underflow to manage; the numerical stability concerns of the batch matcher carry over unchanged.

Formally, lagged smoothing emits the state estimate argmax P(edge_i | z_0 … z_{i+L}) rather than the full-trip posterior argmax P(edge_i | z_0 … z_end). As L grows, the two converge; the whole design is the deliberate choice of a finite L to bound latency.

Routing-Engine Integration

Per-window candidate lookups. Each window needs candidate edges for its fixes. Calling OSRM’s /match or /nearest per fix per window is wasteful because consecutive windows overlap by W - 1 fixes. Cache candidate edges per fix (keyed by rounded (lon, lat) or a geohash tile) so a fix’s candidates are computed once and reused as it slides through successive windows.

Coordinate order. OSRM and Valhalla both expect [longitude, latitude]. A silent lat/lon swap places the query in the wrong hemisphere and returns plausible-looking but entirely wrong edges with no error. Assert axis order at the boundary.

Backpressure. The routing engine, not Kafka, is usually the bottleneck. Cap concurrent engine calls with a semaphore, and when calls queue, pause the Kafka consumer (consumer.pause()) rather than buffering unboundedly in the matcher. Pausing propagates backpressure to the broker cleanly; unbounded in-process buffering ends in an out-of-memory kill. Resume when the engine drains.

Rate limits. A self-hosted engine avoids third-party rate caps entirely, which is why fleet-scale streaming almost always runs a self-hosted OSRM cluster. If you must use a shared endpoint, budget the semaphore to its documented limit and treat 429 responses as backpressure signals.

Operational Troubleshooting

Deployment Checklist

• Source topic keyed by vehicle_id; partition count sized for peak concurrent vehicles
• Window strategy chosen (count vs time) with explicit max_fixes / max_span_s bounds
• Lag horizon L set from measured road-ambiguity length, not guessed
• Emission latency budget verified: L / sampling_rate + processing time < SLA
• Revision rate measured in shadow mode and below tolerance before go-live
• Keyed state backed by changelog or external store; restored on partition assignment
• Idle-vehicle TTL eviction in place so state does not grow unbounded
• Watermark + allowed-lateness configured; too-late fixes routed to a side-output topic
• Offsets committed only after matched edges are produced and state checkpointed
• Routing-engine concurrency capped with a semaphore; consumer pauses on backpressure
• Candidate-edge cache keyed by geohash tile to avoid redundant engine calls
• Coordinate order asserted as [lon, lat] before every routing-engine call
• Cooperative-sticky assignment enabled to minimise partition movement on rebalance

Parent topic: Trajectory Analysis & Map Matching Techniques

Related

• Hidden Markov Model Map Matching in Python — the batch Viterbi decoder this pipeline adapts to bounded windows
• Kafka Consumer Windowed Map Matching in Python — the consumer class that keeps per-vehicle windows and commits offsets after emit
• Exactly-Once Semantics for Streaming Trajectory Matching — Kafka transactions so each matched segment is emitted exactly once
• OSRM Integration for Fleet Map Matching — candidate-edge lookups and per-window routing calls behind the matcher
• Kalman Filtering for GPS Noise Reduction — upstream smoothing that sharpens emission probabilities and lowers the revision rate