Real-Time AIS Vessel Tracking Pipeline on Serverless

A production AIS vessel tracking pipeline ingests raw NMEA !AIVDM sentences into a streaming primitive — Kinesis Data Streams on AWS or Pub/Sub on GCP — then fans them out to Lambda or Cloud Function consumers that decode positions, run point-in-polygon zone checks with shapely, and persist geohashed points to a spatial store. The core design rule is: partition every record by vessel MMSI so per-vessel ordering is preserved end to end, and process each shard with a checkpointed consumer that can sustain 1,000+ position reports per second at sub-second latency. This recipe belongs to the broader family of event-driven geospatial processing patterns, applied here to an unbounded, latency-sensitive stream rather than a batch of files.

Why Streaming AIS Tracking Matters for Geospatial Workloads

AIS (Automatic Identification System) transponders broadcast a vessel’s position, course, speed over ground (SOG), and identity several times per minute. A regional coastal feed easily produces several thousand position reports per second, and a global satellite-AIS feed produces far more. This is a fundamentally different workload from processing a dropped shapefile or a nightly raster batch: the data never stops, ordering matters per vessel, and a stalled consumer means the live map goes stale within seconds.

The decision between streaming and batch is the first architectural fork. If your product answers “where is this vessel right now” or “which vessels just entered this exclusion zone,” you need a streaming path. If it answers “what was the average traffic density last quarter,” a batch path over object storage is cheaper. The tradeoffs are laid out in detail in batch vs stream geospatial processing, and the specific case for AIS — where a few seconds of latency separates a useful collision-avoidance alert from a useless one — is covered in when to use batch vs streaming for real-time AIS tracking.

The geospatial work inside each consumer is deceptively light per message but heavy in aggregate. Decoding an AIVDM payload is a bit-unpacking exercise; the expensive part is the point-in-polygon test against zone geometries (ports, traffic separation schemes, marine protected areas). Loading those polygons and building a spatial index at cold start is where the cold start behavior of Python GDAL and shapely directly affects your per-shard throughput ceiling.

The diagram below shows the full data flow from transponder to spatial store:

Platform-by-Platform Limits

The streaming primitive and its consumer runtime are chosen together. These are the constraints that decide how many vessels and messages per second a single deployment can sustain.

Kinesis binds consumer parallelism directly to shard count: one shard maps to at most one concurrent Lambda per consumer (with standard iterators), so you scale by resharding. Pub/Sub decouples this — it autoscales consumers independently of any partition count, but you must set an ordering key to get the per-MMSI guarantee Kinesis gives for free through the partition key. Azure Functions on the Consumption plan is the most constrained target here; its 10-minute timeout and 1.5 GB memory ceiling make it suitable for lightweight decode-and-forward consumers but not for heavy in-function spatial joins.

Step-by-Step Implementation

Step 1: Ingest AIS NMEA Sentences with a Partition Key of MMSI

The producer reads AIS off a TCP feed, extracts the MMSI from each sentence, and puts the raw payload onto the stream keyed by MMSI. Keying by MMSI is the single most important decision in the pipeline — it is what keeps a vessel’s reports ordered.

# producer.py — forward raw AIS sentences onto Kinesis, keyed by MMSI
import boto3
from pyais import decode  # lightweight AIVDM decoder

kinesis = boto3.client("kinesis")
STREAM = "ais-positions"

def forward(nmea_line: str) -> None:
    # Decode only far enough to read the MMSI for partitioning.
    # Full decoding happens in the consumer to keep the producer thin.
    try:
        msg = decode(nmea_line)
    except Exception:
        return  # skip fragments we cannot key; multipart handled upstream

    mmsi = str(msg.mmsi)
    kinesis.put_record(
        StreamName=STREAM,
        Data=nmea_line.encode("utf-8"),
        PartitionKey=mmsi,  # all reports for one vessel land on one shard
    )

On GCP, the equivalent uses the Pub/Sub publisher with enable_message_ordering=True and an ordering_key=mmsi. The routing tradeoffs between a single stream and per-priority streams are discussed in SQS and Pub/Sub queue routing strategies.

Step 2: Decode AIVDM Payloads in the Consumer

The consumer receives a batch of records, decodes each AIVDM payload into a structured position, and skips non-positional message types. Wrap every record so one malformed sentence cannot fail the batch.

# consumer.py — Lambda handler bound to the Kinesis stream
import base64
import json
from pyais import decode

# Position report message types carry lat/lon/SOG/COG
POSITION_TYPES = {1, 2, 3, 18, 19}

def parse_records(event):
    for rec in event["Records"]:
        seq = rec["kinesis"]["sequenceNumber"]
        raw = base64.b64decode(rec["kinesis"]["data"]).decode("utf-8")
        try:
            msg = decode(raw)
            if msg.msg_type not in POSITION_TYPES:
                continue
            yield seq, {
                "mmsi": msg.mmsi,
                "lat": float(msg.lat),
                "lon": float(msg.lon),
                "sog": float(getattr(msg, "speed", 0.0)),  # knots
                "cog": float(getattr(msg, "course", 0.0)),
                "ts": rec["kinesis"]["approximateArrivalTimestamp"],
            }
        except Exception as exc:
            # Surface the sequence number so we can report it as failed
            yield seq, {"_error": str(exc), "_raw": raw}

Step 3: Run Point-in-Polygon Zone Checks with shapely

Load your zone polygons once at module scope (cold start) and index them with a shapely.STRtree so each position is a fast bounding-box lookup followed by an exact contains test. Building the index per invocation would dominate latency.

# zones.py — module-scope so it survives across warm invocations
import json
from shapely.geometry import shape, Point
from shapely import STRtree

with open("zones.geojson") as f:
    _features = json.load(f)["features"]

_polys = [shape(f["geometry"]) for f in _features]
_names = [f["properties"]["name"] for f in _features]
_tree = STRtree(_polys)  # built once at cold start

def zones_for(lon: float, lat: float) -> list[str]:
    p = Point(lon, lat)
    hits = []
    # STRtree returns candidate indices by bounding box; confirm with contains
    for idx in _tree.query(p):
        if _polys[idx].contains(p):
            hits.append(_names[idx])
    return hits

Keeping the polygon set and index in module scope is the same warm-reuse pattern that makes provisioned concurrency effective; see reducing Python GDAL cold starts with provisioned concurrency for pinning consumers warm during peak traffic.

Step 4: Write Geohashed Positions to the Spatial Store

Encode each position as a geohash and write it to DynamoDB with the MMSI as partition key. The geohash sort key enables both “latest position per vessel” reads and prefix-scan proximity queries.

# store.py — write latest position keyed by MMSI + geohash
import boto3
import pygeohash as pgh

_table = boto3.resource("dynamodb").Table("vessel_positions")

def write_position(pos: dict, zones: list[str]) -> None:
    gh = pgh.encode(pos["lat"], pos["lon"], precision=7)  # ~150 m cells
    _table.put_item(Item={
        "mmsi": pos["mmsi"],           # partition key
        "geohash": gh,                 # sort key
        "lat": str(pos["lat"]),
        "lon": str(pos["lon"]),
        "sog": str(pos["sog"]),
        "zones": zones,
        "ts": int(pos["ts"]),
    })

For the analytical layer, stream the same records to BigQuery GIS via a Firehose-equivalent or a direct insert_rows_json, storing geometry as ST_GEOGPOINT(lon, lat) so you can later run ST_DWithin and ST_Intersects joins across historical tracks. Contrast this discrete write path with the file-batched approach in chunked I/O for large satellite imagery — AIS writes are tiny and frequent, so the store, not the payload, is the bottleneck.

Step 5: Report Partial Failures and Route Poison Records

Assemble the handler so successfully processed sequence numbers are checkpointed and only genuinely failed records are re-delivered or dead-lettered.

# handler.py — glue with partial-batch failure reporting
from consumer import parse_records
from zones import zones_for
from store import write_position

def handler(event, context):
    failures = []
    for seq, pos in parse_records(event):
        if "_error" in pos:
            failures.append({"itemIdentifier": seq})  # retry / DLQ this one
            continue
        try:
            hits = zones_for(pos["lon"], pos["lat"])
            write_position(pos, hits)
        except Exception:
            failures.append({"itemIdentifier": seq})
    # Kinesis re-delivers only these sequence numbers, not the whole batch
    return {"batchItemFailures": failures}

Configure the event source mapping with FunctionResponseTypes=["ReportBatchItemFailures"], a bisect-on-error setting, and a maximum retry count, after which records spill to a dead-letter queue. The full DLQ design — redrive policy, poison detection, and replay — is covered in implementing dead-letter queues for failed vector jobs.

Measurement and Verification

Validate ordering and latency before going live. Emit the arrival-to-store latency per record and confirm per-MMSI monotonic ordering with a synthetic replay.

# verify_latency.py — replay a captured feed and measure end-to-end latency
import time, boto3

logs = boto3.client("logs")
# Query CloudWatch for the custom EMF metric emitted by the handler
resp = logs.start_query(
    logGroupName="/aws/lambda/ais-consumer",
    startTime=int(time.time()) - 900,
    endTime=int(time.time()),
    queryString="fields storeLatencyMs | stats avg(storeLatencyMs), pct(storeLatencyMs, 99)",
)

Expected steady-state numbers for a single-shard consumer at ~800 records/s on a 1,024 MB Lambda:

{
  "avg_store_latency_ms": 42,
  "p99_store_latency_ms": 180,
  "iterator_age_ms": 250,
  "records_per_second": 800
}

The most important health metric is GetRecords.IteratorAgeMilliseconds (Kinesis) or subscription backlog (Pub/Sub). A steadily rising iterator age means the consumer cannot keep up and you must add shards or raise consumer memory. An iterator age near zero with p99 latency under a few hundred milliseconds is a healthy live pipeline.

Failure Modes and Debugging

Iterator age climbing without bound: The consumer is slower than the ingest rate. Each Kinesis shard is served by one concurrent Lambda, so a single overloaded shard cannot be helped by raising account concurrency — you must reshard (split hot shards) or move the expensive zone check to a lighter algorithm. Confirm with the IteratorAgeMilliseconds metric per shard.

All reports for one vessel arriving out of order: The producer used a non-MMSI partition key (a timestamp or random value), scattering one vessel across shards. Fix the PartitionKey to the MMSI and, on Pub/Sub, set ordering_key plus enable_message_ordering.

One malformed sentence stalls a shard forever: Without ReportBatchItemFailures, a record that always throws causes Kinesis to retry the whole batch until it expires, blocking the shard. Enable partial-batch responses and set a maximum record age / retry count with a DLQ target.

Zone checks slow and CPU-bound: The STRtree is being rebuilt every invocation instead of at module scope, or the polygon set is enormous. Move index construction to import time, simplify polygons with shapely.simplify, and consider pre-filtering by geohash cell before the exact contains test.

Multipart AIVDM sentences dropped: Type 5 and long messages span multiple NMEA fragments. If the producer keys fragments independently they can split across shards and never reassemble. Reassemble multipart sentences in the producer before keying, or key all fragments of a group by MMSI.

Cost and Scaling Considerations

Kinesis bills per shard-hour plus per million PUT payload units, so cost scales with provisioned shard count rather than actual message volume — right-size shards to your peak records-per-second and use on-demand mode if traffic is spiky. Lambda cost is dominated by invocation count and duration; batching up to several hundred records per invocation amortizes the per-invocation overhead and keeps the per-message cost in the microdollar range. Because the zone-check CPU work scales with memory, the memory and CPU allocation model for raster workloads applies directly: more memory buys more vCPU, which shortens the shapely pass and can lower total cost by processing batches faster.

Pub/Sub’s autoscaling model removes shard math but shifts cost to per-message delivery and to the consumers Cloud Run or Cloud Functions spins up — watch concurrent instance count during traffic spikes. For the analytical layer, BigQuery streaming inserts are billed per byte ingested; if you do not need per-second freshness in the warehouse, micro-batch positions into files and load them, which is cheaper. The windowed aggregation of AIS positions with Kinesis recipe shows how to collapse thousands of raw reports into per-cell summaries, cutting downstream storage and query cost by one to two orders of magnitude.

Frequently Asked Questions

Why use Kinesis or Pub/Sub instead of an S3 event trigger for AIS data?

AIS feeds are continuous, unbounded, and ordered per vessel — thousands of position reports per second over TCP. A streaming primitive like Kinesis Data Streams or Pub/Sub preserves per-shard ordering, buffers bursts, and lets many consumers process shards in parallel with checkpointing. An S3 event trigger only fires on discrete object writes, which forces you to batch AIS into files first and adds minutes of latency — fine for backfill, wrong for live tracking.

How do I keep positions for the same vessel in order?

Use the vessel MMSI as the partition key when you put records onto the stream. Kinesis and Pub/Sub ordering keys guarantee that all records for one MMSI land on the same shard and are delivered in arrival order. Never use a random or timestamp partition key for AIS, or a single vessel’s reports scatter across shards and arrive out of order.

What spatial store should I use for live vessel positions?

For low-latency lookups of the latest position per vessel, DynamoDB with an MMSI partition key and a geohash sort key gives single-digit-millisecond reads and proximity scans by geohash prefix. For historical analytics and heavy spatial joins across millions of positions, stream into BigQuery GIS and query with ST_ functions. Many pipelines write to both: DynamoDB for the live layer, BigQuery for analytics.

How do I handle a Lambda that fails to decode a malformed AIS message?

Do not let one bad NMEA sentence poison the whole batch. Wrap per-record decoding in a try/except and use ReportBatchItemFailures so Kinesis re-delivers only the specific failed sequence numbers rather than the entire batch. Route records that fail repeatedly to a dead-letter queue for offline inspection instead of blocking the shard iterator.

Related

• Windowed Aggregation of AIS Positions with Kinesis — collapse raw position reports into per-cell vessel counts and average SOG
• Batch vs Stream Geospatial Processing — choose the right processing model for continuous versus bounded geospatial data
• When to Use Batch vs Streaming for Real-Time AIS Tracking — the latency thresholds that make streaming worth its overhead for AIS
• SQS and Pub/Sub Queue Routing Strategies — route AIS records by priority and vessel class across queues
• Implementing Dead-Letter Queues for Failed Vector Jobs — capture and replay poison AIS records without blocking the shard

Back to Serverless Geospatial Pipeline Recipes