Windowed Aggregation of AIS Positions with Kinesis

Windowed AIS aggregation assigns each vessel position to a geohash cell and an event-time window, then maintains a per-(cell, window) aggregate — vessel count and average SOG — in DynamoDB using atomic ADD updates. Use tumbling windows for periodic snapshots and sliding windows for rolling metrics, bucket by the report’s own timestamp rather than arrival time, and hold windows open until a watermark (max event time minus an allowed-lateness grace period) passes their end so out-of-order reports still land in the right bucket.

Event-time windowed aggregation of AIS positions Diagram showing a stream of AIS reports assigned to three consecutive tumbling windows by event time, each window aggregating vessel count and average SOG per geohash cell, with a watermark line marking allowed lateness and a late report routed to a side-output. Event-time axis, tumbling 60 s windows keyed by geohash cell Window W0 [00:00-01:00) cell u4pruy: count 12 avg SOG 9.4 kn Window W1 [01:00-02:00) cell u4pruy: count 15 avg SOG 8.1 kn Window W2 [02:00-03:00) open, accumulating count 7 so far watermark late report event time in W0 reports past the watermark that miss their window go to a late side-output

Context

The real-time AIS vessel tracking pipeline writes one row per raw position report, which is perfect for “where is this vessel now” but overwhelming for questions like “how many vessels are transiting each grid cell per minute” or “what is the average speed in the approach channel.” Aggregation answers those questions and, as a bonus, collapses thousands of raw reports into a handful of per-cell summaries, cutting downstream storage and query cost dramatically. This is the streaming counterpart to the file-batched summarization discussed in batch vs stream geospatial processing: here the window is the unit of work, not the file.

The hard part is time. AIS reports do not arrive in perfect order — satellite relay, TCP buffering, and multi-source fusion all reorder them by seconds to minutes. If you bucket by processing time (when the record reached your consumer) your counts smear across window boundaries. Bucketing by event time (the report’s own timestamp) fixes the counts but forces you to decide how long to wait for stragglers, which is exactly what a watermark encodes.

Prerequisites

Before deploying the aggregator, confirm the following:

  • Runtime: Python 3.11 on AWS Lambda, bound to the Kinesis stream via an event source mapping with ReportBatchItemFailures enabled
  • IAM: kinesis:GetRecords, kinesis:GetShardIterator, kinesis:DescribeStream on the source stream; dynamodb:UpdateItem and dynamodb:Query on the aggregate table
  • State table: DynamoDB ais_window_agg with partition key cell (geohash string) and sort key window_start (epoch seconds), plus numeric attributes count and sog_sum
  • Dependencies: pygeohash==1.2.0 for cell encoding; no native GDAL is required, so the layer stays small and cold starts stay short
  • Environment variables:
    code 
    WINDOW_SECONDS=60          # tumbling window width
ALLOWED_LATENESS_SECONDS=120   # grace period before a window is final
GEOHASH_PRECISION=6        # ~1.2 km cells for density; raise for finer grids
AGG_TABLE=ais_window_agg
  • A working understanding of the per-MMSI ordering guarantees from the parent pipeline — ordering is preserved per shard, but reports from different vessels still interleave.

Implementation

The handler assigns each report to a (cell, window_start) key, merges partial aggregates within the batch to minimize DynamoDB calls, and applies the watermark to decide which reports are too late to count. State lives in DynamoDB via atomic ADD, so concurrent shard consumers never clobber each other’s counts.

python 
# aggregator.py — event-time windowed AIS aggregation on Kinesis + Lambda
import base64
import os
from collections import defaultdict

import boto3
import pygeohash as pgh
from pyais import decode

WINDOW = int(os.environ["WINDOW_SECONDS"])
LATENESS = int(os.environ["ALLOWED_LATENESS_SECONDS"])
PRECISION = int(os.environ["GEOHASH_PRECISION"])
POSITION_TYPES = {1, 2, 3, 18, 19}

_table = boto3.resource("dynamodb").Table(os.environ["AGG_TABLE"])


def window_start(ts: int) -> int:
    """Floor an epoch-second timestamp to its tumbling window boundary."""
    return ts - (ts % WINDOW)


def handler(event, context):
    # Track the max event time in this batch to advance the watermark.
    max_event_ts = 0
    # Partial aggregates merged in-memory before the durable write:
    #   (cell, window_start) -> [count, sog_sum]
    partials: dict[tuple[str, int], list] = defaultdict(lambda: [0, 0.0])
    failures = []
    late_side_output = []

    for rec in event["Records"]:
        seq = rec["kinesis"]["sequenceNumber"]
        try:
            raw = base64.b64decode(rec["kinesis"]["data"]).decode("utf-8")
            msg = decode(raw)
            if msg.msg_type not in POSITION_TYPES:
                continue

            # Prefer the report's own event time; fall back to arrival time.
            ts = int(getattr(msg, "epoch", rec["kinesis"]["approximateArrivalTimestamp"]))
            max_event_ts = max(max_event_ts, ts)

            cell = pgh.encode(float(msg.lat), float(msg.lon), precision=PRECISION)
            w_start = window_start(ts)
            sog = float(getattr(msg, "speed", 0.0))

            agg = partials[(cell, w_start)]
            agg[0] += 1       # vessel report count
            agg[1] += sog     # running SOG sum -> average on read
        except Exception as exc:
            failures.append({"itemIdentifier": seq})

    # Watermark: the frontier of event time we consider "complete".
    watermark = max_event_ts - LATENESS

    for (cell, w_start), (count, sog_sum) in partials.items():
        window_end = w_start + WINDOW
        if window_end <= watermark:
            # Window has already been closed and flushed by a prior batch;
            # this report is past the allowed lateness -> side-output.
            late_side_output.append({"cell": cell, "window_start": w_start, "count": count})
            continue

        # Atomic merge into durable state; safe across concurrent shard workers.
        _table.update_item(
            Key={"cell": cell, "window_start": w_start},
            UpdateExpression="ADD #c :c, sog_sum :s",
            ExpressionAttributeNames={"#c": "count"},
            ExpressionAttributeValues={":c": count, ":s": _dec(sog_sum)},
        )

    if late_side_output:
        _emit_late(late_side_output)  # e.g. to a reconciliation queue

    # Only genuinely un-decodable records are retried / dead-lettered.
    return {"batchItemFailures": failures}


def _dec(x: float):
    from decimal import Decimal
    return Decimal(str(round(x, 3)))  # DynamoDB requires Decimal, not float


def _emit_late(items: list) -> None:
    # Route late reports somewhere for optional backfill; do not block the shard.
    pass

To read a closed window’s average SOG, query by cell for the window_start and compute sog_sum / count at read time — storing the sum rather than a running average keeps the atomic ADD correct under concurrency. For sliding windows, emit each report into every overlapping window key (for a 15-minute window sliding every minute, that is 15 keys) inside the same loop; the state and write volume grow by the overlap factor, so reserve sliding windows for metrics that truly need continuous refresh.

For workloads where hand-rolled windowing becomes unwieldy — session windows, complex event-time triggers, or exactly-once sinks — move the aggregation to Kinesis Data Analytics (Amazon Managed Service for Apache Flink), which provides watermarks, keyed state, and windowing as first-class runtime features. The Lambda approach here is the right choice when the aggregation is simple counts and sums and you want to stay fully serverless without a Flink cluster. The upstream deduplication and idempotency concerns are the same ones covered in deduplicating S3 event notifications for idempotent ingestion, since a redelivered Kinesis record would otherwise double-count.

Verification

Replay a fixed set of reports with known timestamps and confirm the per-cell counts land in the expected windows, including a deliberately late report.

python 
# verify_windows.py — assert counts per (cell, window) after a replay
import boto3

table = boto3.resource("dynamodb").Table("ais_window_agg")
resp = table.query(
    KeyConditionExpression="cell = :c",
    ExpressionAttributeValues={":c": "u4pruy"},
)
for item in resp["Items"]:
    avg_sog = float(item["sog_sum"]) / int(item["count"])
    print(f"window {item['window_start']}: count={item['count']} avg_sog={avg_sog:.1f}")

Expected output for a replay of the diagram’s scenario (a late report correctly counted into window W0):

code 
window 0:  count=12 avg_sog=9.4
window 60: count=15 avg_sog=8.1
window 120: count=7 avg_sog=10.2

If the late report instead inflates the current window’s count, your handler is bucketing by processing time — switch the timestamp source to the report’s event time.

Gotchas and Edge Cases

  • DynamoDB rejects float. Numeric attributes must be Decimal. Passing a Python float to update_item raises TypeError: Float types are not supported. Convert SOG sums with Decimal(str(value)), never Decimal(value), to avoid binary-float precision artifacts.
  • Watermark stalls when the feed goes quiet. The watermark is derived from the max event time seen; if a shard receives no records, its watermark never advances and windows never close. Add a periodic tick (an EventBridge-scheduled invocation) that advances the watermark using wall-clock time so idle cells still finalize.
  • Sliding windows multiply state and cost. A 15-minute window sliding every minute writes each report into 15 keys. Confirm the overlap factor is worth it; if consumers only ever read the latest complete window, a tumbling window plus a small rolling read is far cheaper.
  • Redelivery double-counts. Kinesis guarantees at-least-once delivery, so a retried batch re-adds counts. Make writes idempotent by keying on a per-batch dedupe token, or accept small over-counts on retries if the metric is a density estimate rather than an exact tally.

Frequently Asked Questions

Should I use tumbling or sliding windows for AIS traffic density?

Use tumbling (non-overlapping) windows for periodic snapshots like a per-minute vessel count per cell — each report contributes to exactly one window, which keeps the math and storage simple. Use sliding windows when you need a continuously updated rolling metric, such as a 15-minute average SOG refreshed every minute; sliding windows overlap, so each report contributes to several windows and both compute and state grow with the overlap factor.

How do I handle AIS reports that arrive out of order?

Aggregate by event time — the report’s own timestamp — not processing time. Floor each report to its window boundary and keep windows open until a watermark (the max event time seen minus an allowed-lateness grace period) passes their end. Reports arriving after the watermark closes their window are past the allowed lateness and are dropped or routed to a late side-output for reconciliation.

Where do I keep window state between Lambda invocations?

Lambda is stateless across invocations, so window aggregates must live in an external store. Use a DynamoDB table keyed by (geohash_cell, window_start) and update counts and running SOG sums with atomic ADD expressions. For heavier windowing with built-in state management, use Kinesis Data Analytics (Managed Service for Apache Flink), which keeps window state in the Flink runtime.

Back to Real-Time AIS Vessel Tracking Pipeline