Deduplicating AIS Messages in Real-Time Streams

Problem Framing #

The same vessel position lands in your topic three times a second and every count downstream is wrong. This is the defining nuisance of the Real-Time AIS Stream Ingestion Pipelines workflow: a ship transmits one Type 1 position report, but two shore receivers with overlapping coverage both hear it, a satellite AIS provider relays a third copy, and the Kafka or MQTT transport re-delivers at-least-once after a rebalance or reconnect. The payload is byte-identical or nearly so, yet vessel-density heatmaps inflate, port-call counts double, and dwell-time analytics skew high. This page defines a dedup key that survives multi-receiver fan-in, implements it as a memory-bounded seen-set, and pushes idempotency all the way to the write so a replay cannot resurrect a duplicate you already dropped.

The diagram traces one physical transmission through three ingestion paths that converge on the dedup stage. The key insight: duplicates are not a bug in any single receiver — they are the expected consequence of redundant coverage, and the pipeline must be built to expect them.

Root Cause: Where the Duplicates Come From #

Three mechanisms generate duplicates, and each needs the same key but a different window:

1. Multi-receiver fan-in. Two terrestrial stations that both hear a vessel forward the identical NMEA payload; the only difference is a receiver-added rx_id and possibly an arrival timestamp a few milliseconds apart.
2. Terrestrial + satellite overlap. A satellite provider relays the same report the shore network already delivered, but with 30–90 seconds of backhaul latency, so the two copies can straddle a time window boundary.
3. Transport redelivery. Kafka at-least-once semantics and MQTT QoS1 both replay on reconnect or rebalance — the exact scenario produced by the fixes in troubleshooting AIS Kafka & MQTT connector failures.

A raw duplicate pair looks like this — identical except the receiver id:

# Two rows for ONE physical position report
{"mmsi": 235098765, "timestamp": "2026-07-13T09:14:07Z", "lat": 50.80213, "lon": -1.08841, "rx_id": "SHORE-A"}
{"mmsi": 235098765, "timestamp": "2026-07-13T09:14:07Z", "lat": 50.80213, "lon": -1.08841, "rx_id": "SHORE-B"}

The dedup key must therefore exclude rx_id and any receiver-local metadata, and include the identity of the position: mmsi, the reported timestamp, and the position rounded to a tolerance that absorbs sub-metre float noise. Rounding latitude/longitude to five decimal places (~1.1 m at the equator) collapses receivers that decoded the same bits into slightly different floats without merging genuinely distinct positions.

Step-by-Step Fix #

Step 1 — Define the dedup key #

from __future__ import annotations

import logging
from typing import Any

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s — %(message)s")
logger = logging.getLogger("ais.dedup")

# ~1.1 m at the equator; absorbs float-decode jitter, keeps distinct fixes distinct
_POS_DECIMALS = 5


def dedup_key(msg: dict[str, Any]) -> tuple[int, str, float, float]:
    """Build a receiver-independent identity for one AIS position report.

    Raises KeyError if a required field is missing so a malformed record
    cannot silently collapse into another vessel's key.
    """
    try:
        mmsi = int(msg["mmsi"])
        ts = str(msg["timestamp"])
        lat = round(float(msg["lat"]), _POS_DECIMALS)
        lon = round(float(msg["lon"]), _POS_DECIMALS)
    except (KeyError, TypeError, ValueError) as exc:
        raise KeyError(f"cannot build dedup key from {msg!r}: {exc}") from exc
    return (mmsi, ts, lat, lon)

Step 2 — A bounded TTL seen-set #

An unbounded set of keys will exhaust memory on a busy feed within hours. The seen-set must evict by time, not just by count, because a satellite duplicate can arrive 90 seconds after the terrestrial original. The implementation below keeps insertion order via a plain dict (ordered since Python 3.7) and evicts entries older than the TTL on every insert. It raises if the structure grows past a hard ceiling — a growth runaway means the TTL is misconfigured or the clock is broken, and silently unbounded memory is worse than a loud crash:

import time


class TtlSeenSet:
    """Time-bounded membership test for dedup keys.

    Keys expire after `ttl_s`. A hard `max_size` ceiling guards against
    unbounded growth (misconfigured TTL, clock skew) by raising rather than
    leaking memory until the pod is OOM-killed.
    """

    def __init__(self, ttl_s: float = 120.0, max_size: int = 5_000_000) -> None:
        if ttl_s <= 0:
            raise ValueError("ttl_s must be positive")
        self._ttl_s = ttl_s
        self._max_size = max_size
        self._seen: dict[tuple, float] = {}

    def _evict_expired(self, now: float) -> None:
        cutoff = now - self._ttl_s
        # dict preserves insertion order; oldest keys are at the front
        stale = 0
        for key, inserted in self._seen.items():
            if inserted >= cutoff:
                break
            stale += 1
        for key in list(self._seen.keys())[:stale]:
            del self._seen[key]

    def seen_before(self, key: tuple) -> bool:
        """Return True if `key` is a duplicate within the TTL window; else record it."""
        now = time.monotonic()
        self._evict_expired(now)
        if key in self._seen:
            return True
        self._seen[key] = now
        if len(self._seen) > self._max_size:
            raise MemoryError(
                f"seen-set exceeded {self._max_size} keys — TTL {self._ttl_s}s too long "
                "or upstream clock skew is inflating unique keys"
            )
        return False

Step 3 — Wire it into the poll loop #

def filter_duplicates(
    records: list[dict[str, Any]],
    seen: TtlSeenSet,
) -> list[dict[str, Any]]:
    """Drop exact duplicates; log the duplicate ratio for observability."""
    kept: list[dict[str, Any]] = []
    dupes = 0
    for rec in records:
        key = dedup_key(rec)
        if seen.seen_before(key):
            dupes += 1
            continue
        kept.append(rec)
    if records:
        logger.info("dedup: kept %d, dropped %d (%.1f%% dup)",
                    len(kept), dupes, 100.0 * dupes / len(records))
    return kept

Step 4 — Near-duplicate collapse and idempotent writes #

Exact-key dedup misses a subtle case: two receivers that report the same fix one second apart (their clocks differ) produce two keys. Collapse near-duplicates with a short per-MMSI window using pandas, keeping the first arrival:

import pandas as pd
import numpy as np


def collapse_near_duplicates(df: pd.DataFrame, window_s: float = 2.0) -> pd.DataFrame:
    """Within each MMSI, drop rows whose position matches the previous kept row
    and fall inside `window_s`. Assumes df is timestamp-sorted per MMSI."""
    if not np.issubdtype(df["timestamp"].dtype, np.datetime64):
        raise TypeError("timestamp column must be datetime64 before collapsing")
    out: list[pd.DataFrame] = []
    for mmsi, grp in df.sort_values("timestamp").groupby("mmsi", sort=False):
        grp = grp.copy()
        dt = grp["timestamp"].diff().dt.total_seconds().fillna(np.inf)
        same_pos = (
            grp["lat"].round(_POS_DECIMALS).diff().fillna(1).eq(0)
            & grp["lon"].round(_POS_DECIMALS).diff().fillna(1).eq(0)
        )
        drop = same_pos & (dt <= window_s)
        out.append(grp.loc[~drop])
    return pd.concat(out, ignore_index=True)

Finally, make the downstream write idempotent so a replayed batch cannot reintroduce a row you already dropped: use the dedup key as a primary/merge key. For a warehouse, MERGE ... ON (mmsi, timestamp, lat, lon); for object storage, name Parquet objects deterministically from the batch’s key range so a re-write overwrites rather than appends.

Verification #

Assert the duplicate ratio falls and the seen-set stays bounded on a replayed fixture:

def test_dedup_drops_duplicates() -> None:
    seen = TtlSeenSet(ttl_s=120.0)
    base = {"mmsi": 235098765, "timestamp": "2026-07-13T09:14:07Z", "lat": 50.80213, "lon": -1.08841}
    stream = [
        {**base, "rx_id": "SHORE-A"},
        {**base, "rx_id": "SHORE-B"},   # duplicate
        {**base, "rx_id": "SAT-3"},     # duplicate
        {**base, "timestamp": "2026-07-13T09:14:09Z", "rx_id": "SHORE-A"},  # new fix
    ]
    kept = filter_duplicates(stream, seen)
    unique_keys = {dedup_key(r) for r in kept}
    assert len(kept) == 2, f"expected 2 unique, got {len(kept)}"
    assert len(unique_keys) == len(kept), "kept set still contains duplicates"
    print("PASS — 4 messages collapsed to 2 unique positions")

Edge Cases and Gotchas #

• Legitimate identical-timestamp messages. A vessel’s Type 5 static/voyage report and its Type 1 position can share a second-of-minute timestamp yet carry different payloads. Because the dedup key includes lat/lon (absent or sentinel on Type 5), do not extend the key to fields that differ by message type, and never dedup across message classes on timestamp alone.
• Clock skew makes duplicates look distinct. If one receiver’s NTP is wrong, the same physical fix carries two different timestamps and slips past the key — the exact failure covered in handling AIS timestamp drift & clock skew. Normalise timestamps to a corrected reference before dedup, or the seen-set inflates and eventually trips its MemoryError ceiling.
• Memory bound of the seen-set. The seen-set size is roughly peak_msg_rate × ttl_s × unique_fraction. At 10 k msg/s and a 120 s TTL, that is up to 1.2 M keys — comfortably under the 5 M ceiling, but push the TTL to 900 s and you approach it. Size max_size from your measured rate, and export len(seen._seen) as a gauge so you see it climbing before it raises. Residual gaps from over-aggressive dedup should be reconciled in AIS data quality & gap filling.

---

Related #

• Troubleshooting AIS Kafka & MQTT Connector Failures — the reconnect replays that make dedup mandatory in the first place
• Handling AIS Timestamp Drift & Clock Skew — normalise timestamps before keying dedup on them
• AIS Data Quality & Gap Filling — reconcile gaps that dedup may expose or create

Up: Real-Time AIS Stream Ingestion Pipelines