Every technique in the geospatial canon assumes the data holds still. Spatial indexes are built once and queried many times. Joins run over complete tables. Aggregations scan a dataset that is not changing underneath them.

Moving objects break all of it. Fifty thousand vehicles reporting every five seconds is ten thousand position updates per second, each of which potentially crosses a boundary, enters a restricted zone, exceeds a limit, or stops somewhere it should not. You cannot answer those questions by rebuilding an index, and you cannot answer them by running a spatial join every five seconds either.

This article builds the streaming alternative from open components, and the central trick is to stop doing geometry in the hot path.

# Why the Batch Approach Fails

The obvious implementation is a table of current positions and a periodic spatial join against a zones table. It fails for three separate reasons, and it is worth being precise about which.

Throughput. A ST_Contains join of 50,000 points against 5,000 zone polygons takes a second or two in PostGIS. At a five-second reporting interval you have consumed half your budget doing nothing else, and you have no headroom for growth.

Transitions are invisible. A periodic join tells you where things are. The valuable events are changes — entered, exited, dwelled, deviated. Reconstructing transitions by comparing consecutive snapshots means storing every snapshot and accepting that anything happening between two polls is lost.

Write amplification. Fifty thousand updates per five seconds against an indexed PostGIS table is 10,000 index updates per second on a spatial index that was designed for read-heavy workloads. The index bloats, autovacuum falls behind, and query latency degrades exactly when you need it.

The streaming approach inverts the model: hold the state in the stream processor, evaluate each event once as it arrives, and write only the events — the entries and exits — to the database.

# Ingest

Redpanda is Kafka-API-compatible, written in C++, and requires no ZooKeeper or KRaft configuration. For this workload it is simply less to operate.

docker run -d --name redpanda -p 9092:9092 \
  redpandadata/redpanda:latest redpanda start \
  --overprovisioned --smp 2 --memory 2G --reserve-memory 0M --node-id 0 \
  --check=false --kafka-addr PLAINTEXT://0.0.0.0:9092 \
  --advertise-kafka-addr PLAINTEXT://localhost:9092

rpk topic create positions --partitions 12 --replicas 1
rpk topic create geofence-events --partitions 6 --replicas 1

The partition count is a design decision, not a default. Twelve partitions is twelve-way parallelism downstream, and the key you partition on determines whether your stateful operators can work independently — which is the next section.

Devices usually speak Message Queuing Telemetry Transport (MQTT) rather than the Kafka protocol, so a broker bridges the two:

devices ──MQTT──▶ EMQX ──bridge──▶ Redpanda ──▶ Flink ──▶ TimescaleDB + WebSocket

from confluent_kafka import Producer
import json, time

producer = Producer({"bootstrap.servers": "localhost:9092", "linger.ms": 20,
                     "compression.type": "lz4"})

def emit(vehicle_id, lon, lat, speed_kmh, heading):
    payload = {"vehicle_id": vehicle_id, "lon": lon, "lat": lat,
               "speed_kmh": speed_kmh, "heading": heading,
               "event_ts": int(time.time() * 1000)}
    # Key by vehicle so all events for one vehicle land on one partition,
    # preserving per-vehicle ordering.
    producer.produce("positions", key=vehicle_id, value=json.dumps(payload))

Keying by vehicle guarantees ordering per vehicle, which matters because a transition is defined by comparing an event to its predecessor.

# The H3 Trick

Here is the design decision that makes the whole pipeline cheap.

Instead of testing each position against zone polygons, precompute the set of H3 cells covering every zone, once, offline. At stream time, convert the incoming position to its H3 cell — a pure arithmetic operation on the coordinates, no index, no geometry — and look the cell up in a hash map.

import h3, geopandas as gpd, json

RES = 10   # ~65 m edge; tune to your smallest zone

zones = gpd.read_file("zones.gpkg").to_crs(4326)
cell_to_zones = {}
for zone in zones.itertuples():
    cells = h3.geo_to_cells(zone.geometry, RES)
    for c in cells:
        cell_to_zones.setdefault(c, []).append(zone.zone_id)

print(f"{len(zones)} zones → {len(cell_to_zones)} cells")
with open("cell_index.json", "w") as f:
    json.dump(cell_to_zones, f)

The lookup becomes:

cell = h3.latlng_to_cell(lat, lon, RES)
zone_ids = cell_to_zones.get(cell, [])

A geometry test becomes a dictionary lookup. That is the difference between a spatial join at 10,000 events per second and a hash lookup at 10,000 events per second — roughly three orders of magnitude of headroom.

The approximation is real and you must size it deliberately. Hexagons do not follow zone boundaries, so containment near an edge is approximate to about the cell’s edge length. At resolution 10 that is roughly 65 metres; at resolution 12 it is about 9 metres, at the cost of sixteen times more cells per zone.

Choose by consequence. Geofencing a 50-hectare depot for arrival notifications: resolution 9 is ample. Enforcing a tolling cordon where a false positive costs money and generates a dispute: use the hybrid — H3 for the cheap rejection of everything clearly inside or outside, and an exact ST_Contains only for events landing in cells that straddle a boundary. Those boundary cells are a small fraction of the total, so you keep almost all the performance and get exact answers where it matters.

# Precompute which cells are ambiguous — needed for the hybrid check
boundary_cells = {
    c for c, zs in cell_to_zones.items()
    if not zones[zones.zone_id.isin(zs)].geometry.union_all()
             .contains(shapely.geometry.Polygon(h3.cell_to_boundary(c)))
}

Apache Flink holds per-key state and evaluates each event once. Keying the stream by vehicle means each parallel instance owns a disjoint set of vehicles and never coordinates with the others.

from pyflink.datastream import StreamExecutionEnvironment, RuntimeContext
from pyflink.datastream.functions import KeyedProcessFunction
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.common import Types, WatermarkStrategy, Duration
import h3, json

class GeofenceTransitions(KeyedProcessFunction):
    """Emits entered/exited events by comparing each position to the last one."""

    def open(self, ctx: RuntimeContext):
        self.prev_zones = ctx.get_state(
            ValueStateDescriptor("prev_zones", Types.PICKLED_BYTE_ARRAY()))
        self.entered_at = ctx.get_state(
            ValueStateDescriptor("entered_at", Types.PICKLED_BYTE_ARRAY()))
        with open("cell_index.json") as f:
            self.cell_to_zones = json.load(f)

    def process_element(self, value, ctx):
        ev = json.loads(value)
        cell = h3.latlng_to_cell(ev["lat"], ev["lon"], 10)
        now_zones = set(self.cell_to_zones.get(cell, []))
        was_zones = self.prev_zones.value() or set()
        entered_at = self.entered_at.value() or {}

        for z in now_zones - was_zones:
            entered_at[z] = ev["event_ts"]
            yield json.dumps({"type": "entered", "vehicle_id": ev["vehicle_id"],
                              "zone_id": z, "ts": ev["event_ts"],
                              "lon": ev["lon"], "lat": ev["lat"]})

        for z in was_zones - now_zones:
            dwell_s = (ev["event_ts"] - entered_at.pop(z, ev["event_ts"])) / 1000
            yield json.dumps({"type": "exited", "vehicle_id": ev["vehicle_id"],
                              "zone_id": z, "ts": ev["event_ts"],
                              "dwell_seconds": round(dwell_s, 1)})

        self.prev_zones.update(now_zones)
        self.entered_at.update(entered_at)

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(12)                       # match the partition count

positions = env.add_source(kafka_source("positions")) \
    .assign_timestamps_and_watermarks(
        WatermarkStrategy
          .for_bounded_out_of_orderness(Duration.of_seconds(30))
          .with_timestamp_assigner(lambda e, _: json.loads(e)["event_ts"]))

positions.key_by(lambda e: json.loads(e)["vehicle_id"]) \
         .process(GeofenceTransitions()) \
         .add_sink(kafka_sink("geofence-events"))

env.execute("geofence-transitions")

Two things in that code deserve emphasis.

Dwell time comes free. Because the operator already holds entry timestamps in keyed state, dwell duration is a subtraction at exit rather than a separate analysis. The same state naturally supports “alert if dwell exceeds 20 minutes” using a registered timer.

Watermarks handle late data. Positioning data arrives out of order — devices buffer through tunnels and flush on reconnection. for_bounded_out_of_orderness(30s) tells Flink to wait 30 seconds of event time before considering a window closed. Set this from your observed distribution of arrival lateness, not from a guess: too short and you silently drop late events, too long and every downstream alert inherits the delay.

Duplicate suppression belongs here too. Devices retransmit on unacknowledged sends, so a (vehicle_id, event_ts) seen-set in state with a short expiry prevents one physical movement generating two entry events.

# The Historical Tail

Events go to TimescaleDB, which is PostgreSQL and therefore still has PostGIS available for the analytical queries that are not in the hot path.

CREATE TABLE geofence_events (
  ts            timestamptz NOT NULL,
  vehicle_id    text        NOT NULL,
  zone_id       text        NOT NULL,
  event_type    text        NOT NULL,
  dwell_seconds double precision,
  geom          geometry(Point, 4326)
);
SELECT create_hypertable('geofence_events', 'ts', chunk_time_interval => INTERVAL '1 day');
CREATE INDEX ON geofence_events (zone_id, ts DESC);

-- Rollup the dashboard queries instead of scanning raw events
CREATE MATERIALIZED VIEW zone_activity_hourly
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 hour', ts) AS bucket,
       zone_id,
       count(*) FILTER (WHERE event_type = 'entered') AS entries,
       count(*) FILTER (WHERE event_type = 'exited')  AS exits,
       avg(dwell_seconds) FILTER (WHERE event_type = 'exited') AS mean_dwell_s
FROM geofence_events
GROUP BY bucket, zone_id;

SELECT add_continuous_aggregate_policy('zone_activity_hourly',
  start_offset => INTERVAL '3 days', end_offset => INTERVAL '1 hour',
  schedule_interval => INTERVAL '10 minutes');

-- And do not keep raw events forever
SELECT add_retention_policy('geofence_events', INTERVAL '18 months');

Writing only transitions rather than every position is what makes this sustainable. Fifty thousand vehicles produce roughly 10,000 positions per second but only a few hundred zone transitions per minute. The database sees four orders of magnitude less write traffic than the naive design.

# Getting It to the Screen

Fan-out to browsers is a solved problem that people insist on re-solving. Centrifugo subscribes to the events topic and pushes to WebSocket clients with channel-based authorisation, so a depot manager sees only their own zones:

import { Centrifuge } from 'centrifuge';
import maplibregl from 'maplibre-gl';

const centrifuge = new Centrifuge('wss://events.example.com/connection/websocket',
                                  { token: jwtFromYourAuth });

const sub = centrifuge.newSubscription('zones:depot-14');
sub.on('publication', ({ data }) => {
  // Mutate the existing source rather than re-adding layers
  const src = map.getSource('live-vehicles');
  const fc = src._data;
  const idx = fc.features.findIndex(f => f.properties.vehicle_id === data.vehicle_id);
  const feature = {
    type: 'Feature',
    geometry: { type: 'Point', coordinates: [data.lon, data.lat] },
    properties: { vehicle_id: data.vehicle_id, state: data.type },
  };
  idx >= 0 ? (fc.features[idx] = feature) : fc.features.push(feature);
  src.setData(fc);
});

sub.subscribe();
centrifuge.connect();

The critical detail is on the client: update the source’s data, do not add and remove layers. Adding a layer per update triggers a style recompilation and will drop frames within a minute. This is the most common cause of “the live map is janky” and it is entirely avoidable.

# Sizing It

Fifty thousand vehicles at a five-second interval:

Quantity Value
Ingest rate 10,000 events/s
Payload ~120 bytes JSON, ~45 bytes compressed
Broker throughput ~1.2 MB/s raw
Flink state ~200 bytes/vehicle → ~10 MB total
Zone transitions ~200–600/minute
Database writes ~10/s
Broker retention, 7 days ~120 GB

That is a small deployment. Redpanda on 4 vCPU, Flink with parallelism 12 across two task managers, TimescaleDB on 8 vCPU. The state size is the number worth internalising: 10 MB of keyed state replaces a spatial index that would otherwise be rebuilt continuously.

Scaling to a million objects is roughly linear in ingest and state, provided the partition key stays high-cardinality. Where it stops being linear is if you key by zone instead of by vehicle — a single busy zone then becomes one hot partition and one overloaded operator instance while eleven sit idle. Key by the moving thing, not by the static thing.

# What This Composes With

The transitions this pipeline emits are the live-state layer of a geospatial digital twin — bind zone_id and vehicle_id to the twin’s semantic objects and the twin becomes current rather than historical. The H3 indexing is the same trick applied in H3 as a Spatial Join Accelerator, just evaluated per event instead of per batch. And the deployment shape follows the event-driven patterns in Cloud-Orchestrated Geospatial Workflows.

The thing to take from it is the inversion. Batch spatial processing asks “where is everything right now” and pays to recompute the answer. Streaming asks “what changed” and pays only for the change. For anything that moves, the second question is both cheaper to answer and the one you actually wanted.

Related reading: Building a Geospatial Digital Twin: Imagery, Processing, and Live Data · Cloud-Orchestrated Geospatial Workflows: AWS, GCP, and Azure · H3 as a Spatial Join Accelerator: 100 Million Points Without sjoin