Kafka Consumer Windowed Map Matching in Python

This page implements the consumer at the centre of a streaming map-matching pipeline: a single class that reads GPS fixes from a Kafka topic keyed by vehicle_id, keeps a bounded sliding window per vehicle, runs a windowed map-match on every new fix, emits committed edges with a lag, and — critically — commits its source offset only after those edges are produced. It extends the batch HMM matcher to the unbounded-stream case where you can never see the end of a trip. The specific edge case it handles is offset/emit ordering across a rebalance: get that wrong and a crash either drops matches or double-counts them.

The window and decode logic here is deliberately thin — the parent pipeline page covers the HMM emission/transition model and the lag-horizon math in full. The focus below is the consumer mechanics: keyed state, manual commit, and rebalance-safe assignment.

Compatibility and Configuration Requirements

Tables scroll horizontally on narrow viewports.

The Consumer Class

The class below is complete and runnable in shape against a live broker. It plugs a WindowedMatcher (your HMM decoder from the pipeline page) into a confluent-kafka consume-emit-commit loop. Every configuration choice is commented at the point it matters.

import json
from collections import deque, defaultdict
from confluent_kafka import Consumer, Producer, TopicPartition


class WindowedMatchConsumer:
    """
    Streaming map-matching consumer with per-vehicle sliding windows.

    Reads keyed GPS fixes, maintains a bounded window per vehicle_id, runs a
    windowed matcher on each fix, emits committed edges older than the lag
    horizon, and commits the source offset only AFTER emit.

    Parameters
    ----------
    brokers : str
        Kafka bootstrap servers, e.g. "broker:9092".
    group_id : str
        Consumer group id. All instances sharing it split partitions.
    in_topic / out_topic : str
        Source fixes topic and destination matched-edges topic.
    window_size : int
        Max fixes retained per vehicle. Larger = more context, more CPU/RAM.
        20-60 for 1 Hz urban data; raise for sparse sampling.
    slide : int
        Re-run the matcher only every `slide` new fixes per vehicle to cap
        cost. slide=1 matches on every fix (lowest latency, highest CPU).
    lag : int
        Fixes held back as provisional; only older edges are emitted.
        Set to the local road-ambiguity length. Bigger = fewer revisions.
    max_poll_records : int
        Upper bound on records returned per poll. Keep small enough that one
        batch processes well under max.poll.interval.ms (each fix may hit the
        routing engine).
    """

    def __init__(self, brokers, group_id, in_topic, out_topic,
                 matcher, window_size=40, slide=1, lag=8,
                 max_poll_records=200, session_timeout_ms=45000):
        self.in_topic = in_topic
        self.out_topic = out_topic
        self.matcher = matcher
        self.window_size = window_size
        self.slide = slide
        self.lag = lag
        self.max_poll_records = max_poll_records

        self.consumer = Consumer({
            "bootstrap.servers": brokers,
            "group.id": group_id,
            "enable.auto.commit": False,          # commit manually after emit
            "auto.offset.reset": "latest",        # skip old backlog on cold start
            "partition.assignment.strategy": "cooperative-sticky",
            # session_timeout must exceed worst-case processing stalls or the
            # broker evicts us mid-batch and triggers a needless rebalance
            "session.timeout.ms": session_timeout_ms,
            "max.poll.interval.ms": 300000,
        })
        self.producer = Producer({
            "bootstrap.servers": brokers,
            "linger.ms": 20,
            "enable.idempotence": True,           # dedupe producer retries
        })

        # per-vehicle state: sliding window + fixes-since-last-match counter
        self.windows: dict[str, deque] = defaultdict(
            lambda: deque(maxlen=self.window_size))
        self.since_match: dict[str, int] = defaultdict(int)
        self.emitted_upto: dict[str, float] = defaultdict(lambda: float("-inf"))

    # -- rebalance callbacks -------------------------------------------------

    def _on_assign(self, consumer, partitions):
        """Restore per-vehicle windows for newly assigned partitions."""
        for tp in partitions:
            for vehicle_id, fixes in self._restore_state(tp).items():
                self.windows[vehicle_id] = deque(fixes, maxlen=self.window_size)
        consumer.assign(partitions)

    def _on_revoke(self, consumer, partitions):
        """Flush emits and commit before losing these partitions."""
        self.producer.flush()
        try:
            consumer.commit(asynchronous=False)
        except Exception:
            pass  # revoke may race a rebalance; next owner re-reads uncommitted

    def _restore_state(self, tp: TopicPartition) -> dict:
        """
        Rebuild windows from a changelog/state store for a partition.
        Return {vehicle_id: [fix, ...]}. Empty dict = cold start (acceptable;
        windows simply refill from the live stream).
        """
        return {}  # wire to RocksDB/Redis/changelog in production

    # -- main loop -----------------------------------------------------------

    def run(self):
        self.consumer.subscribe(
            [self.in_topic], on_assign=self._on_assign, on_revoke=self._on_revoke)
        try:
            while True:
                batch = self.consumer.consume(
                    num_messages=self.max_poll_records, timeout=1.0)
                if not batch:
                    continue

                produced = False
                for msg in batch:
                    if msg is None or msg.error():
                        continue
                    vehicle_id = msg.key().decode("utf-8")
                    fix = json.loads(msg.value())
                    produced |= self._handle_fix(vehicle_id, fix)

                # emit before commit: flush the producer, THEN advance offsets
                if produced:
                    self.producer.flush()
                self.consumer.commit(asynchronous=False)
        finally:
            self.producer.flush()
            self.consumer.close()

    def _handle_fix(self, vehicle_id: str, fix: dict) -> bool:
        """Append a fix, maybe match, emit committed edges. Returns True if emitted."""
        window = self.windows[vehicle_id]
        window.append(fix)
        self.since_match[vehicle_id] += 1

        # cap cost: only re-decode every `slide` fixes
        if self.since_match[vehicle_id] < self.slide:
            return False
        self.since_match[vehicle_id] = 0

        if len(window) <= self.lag:
            return False  # not enough context to commit anything yet

        edges = self.matcher.match_window(
            list(window), lag=self.lag,
            emitted_upto=self.emitted_upto[vehicle_id])

        emitted = False
        for edge in edges:
            self.producer.produce(
                self.out_topic,
                key=vehicle_id.encode("utf-8"),
                value=json.dumps(edge).encode("utf-8"))
            self.emitted_upto[vehicle_id] = edge["event_ts"]
            emitted = True
        return emitted

Execution and Tuning Guidelines

Instantiate with a WindowedMatcher that exposes match_window(fixes, lag, emitted_upto) returning a list of committed (vehicle_id, event_ts, edge_id) dicts — the forward-pass and lagged-backtrack code from the pipeline page. Run one process per available CPU up to the partition count; the group protocol splits partitions across them automatically.

• window_size — the context length. Raising it stabilises decoding through long parallel-road stretches at higher per-fix CPU and memory (the deque(maxlen=...) bounds RAM automatically). Lowering it risks flicker where roads run close together. Tie it to the sampling rate: 40 fixes is ~40 s of context at 1 Hz but ~400 s at 0.1 Hz.
• slide — the cost/latency lever unique to streaming. slide=1 re-decodes on every fix for minimum latency but maximum CPU and routing-engine load. Raising it to 3-5 amortises the decode across several fixes; emissions then arrive in small bursts, which is fine as long as slide stays well below lag.
• lag — held-back provisional fixes, identical in meaning to the pipeline’s lag horizon. Raise it until the measured revision rate drops below tolerance; lower it only if the resulting emission latency (lag / sampling_rate) breaches your SLA.
• max_poll_records — bound the poll batch so its worst-case processing time (fixes × per-window decode × possible engine round-trips) stays comfortably under max.poll.interval.ms. If it does not, the broker assumes the consumer is dead and rebalances, which loses in-memory windows. Start at 200 and lower under heavy engine latency.
• session_timeout_ms — must exceed transient processing stalls (GC pauses, slow engine calls) or you get spurious rebalances. 45 s is a safe default; pair it with a max.poll.interval.ms above your worst-case batch time.

Feed the consumer clean input: applying Kalman filtering upstream sharpens emission probabilities so windows converge with a smaller lag. Candidate-edge lookups should hit a self-hosted OSRM instance to avoid third-party rate limits.

Common Pitfalls

Up: Streaming Map-Matching Pipelines | Trajectory Analysis & Map Matching Techniques

Related

• Streaming Map-Matching Pipelines — the full pipeline: windowing strategy, lag-horizon math, watermarks, and backpressure
• Exactly-Once Semantics for Streaming Trajectory Matching — replace idempotent at-least-once with transactional consume-process-produce
• Hidden Markov Model Map Matching in Python — the batch decoder whose forward pass the windowed matcher reuses
• OSRM Integration for Fleet Map Matching — the candidate-edge source behind each windowed match
• Kalman Filtering for GPS Noise Reduction — upstream smoothing that lets windows converge with a smaller lag