Streaming & Chunking Large Geometry Payloads

A geometry that will not fit in one message can be chunked, referenced or streamed, and only referencing keeps the event atomic — chunking turns one event into many that can partially fail, while streaming solves the consumer’s memory problem and does nothing about the broker’s size limit.

This topic sits under Spatial Payload Routing & Parsing, which covers how a spatial payload gets from the wire into a usable geometry. It assumes the geometry has already been normalised as described in CRS Normalization Strategies, and it is the transport half of the problem whose compute half is Async Processing for Geometry-Heavy Payloads — that page keeps a heavy geometry off the event loop; this one gets it there at all.

Prerequisites

• Python 3.11+ with asyncio — the streaming paths below are all async
• ijson 3.2+ — event-driven JSON parsing without materialising the document
• S3-compatible object storage — any store with content-addressed keys and presigned URLs
• shapely 2.x — STRtree and the vectorised predicates used when reassembling
• A measured size distribution for the stream — percentiles, not the mean
• A broker whose limit you know — Kafka max.message.bytes, SQS 256 KB, Redis Streams practical ceiling

Measure the distribution before choosing anything

Geometry size is not normally distributed and its mean is useless. A stream of parcel boundaries is kilobytes at the median and hundreds of megabytes at the maximum, because one record is a national coastline and the rest are back gardens.

import statistics
from pathlib import Path


def size_profile(samples: list[bytes]) -> dict[str, int]:
    """The five numbers that decide the strategy.

    The mean is deliberately absent: with a distribution this skewed it sits
    below the 90th percentile and describes nothing that will actually break.
    """
    sizes = sorted(len(s) for s in samples)
    q = statistics.quantiles(sizes, n=100, method="inclusive")
    return {
        "p50": int(q[49]), "p90": int(q[89]), "p99": int(q[98]),
        "p999": int(q[-1]), "max": sizes[-1],
    }

The decision falls out of two of those numbers. If p99 fits comfortably inside the broker limit, inline the geometry and handle the tail by reference. If p50 is already close to the limit, the stream needs a different transport entirely.

Three strategies, three different failure modes

Architecture: an envelope that decides per event

The envelope is the design. It carries enough spatial metadata to route and filter without the body, and it names where the body is when the body is not inline.

Layer 1 — measure at the producer. Serialise the geometry, take its length, and compare against an inline threshold set well below the broker limit.

Layer 2 — branch. Under threshold, the geometry goes in the envelope. Over it, the geometry goes to object storage under a content-addressed key and the envelope carries the key.

Layer 3 — always carry the bounding box and vertex count. These make the envelope routable and let a consumer estimate cost before fetching anything, which is what Backpressure & Flow Control for Spatial Consumers charges against its budget.

Layer 4 — resolve lazily at the consumer. Fetch the body only after deciding the event is relevant. A consumer filtering on a geofence discards most events without ever paying for their geometry.

Step-by-step implementation

Step 1 — Publish an envelope that branches on size

import hashlib
import json
from dataclasses import dataclass, asdict

from shapely.geometry import shape

INLINE_LIMIT = 256 * 1024      # well under any broker limit, by design


@dataclass(slots=True)
class SpatialEnvelope:
    feature_id: str
    occurred_at: str
    crs: str                    # always explicit — "EPSG:4326"
    bbox: tuple[float, float, float, float]
    vertex_count: int
    geometry: dict | None = None       # inline path
    body_key: str | None = None        # claim-check path
    body_sha256: str | None = None
    body_bytes: int | None = None


async def publish(geometry: dict, feature_id: str, occurred_at: str,
                  store, broker) -> None:
    geom = shape(geometry)
    body = json.dumps(geometry, separators=(",", ":")).encode()

    env = SpatialEnvelope(
        feature_id=feature_id,
        occurred_at=occurred_at,
        crs="EPSG:4326",
        bbox=geom.bounds,
        vertex_count=count_vertices(geom),
    )

    if len(body) < INLINE_LIMIT:
        env.geometry = geometry
    else:
        digest = hashlib.sha256(body).hexdigest()
        # Content-addressed: a retry writes the same bytes to the same key,
        # so the store write is idempotent and needs no coordination.
        env.body_key = f"geom/{digest[:2]}/{digest}.json"
        env.body_sha256 = digest
        env.body_bytes = len(body)
        await store.put(env.body_key, body)

    await broker.send(json.dumps(asdict(env)).encode())

Writing the body before publishing the envelope is not optional. Reversed, a consumer can receive a reference to an object that does not exist yet and fail on a race that only appears under load.

Step 2 — Resolve lazily, and verify what comes back

class BodyMismatch(Exception):
    """The stored body is not the one the envelope described."""


async def resolve(env: SpatialEnvelope, store) -> dict:
    if env.geometry is not None:
        return env.geometry

    body = await store.get(env.body_key)

    # Verify, always. A truncated read parses into a smaller but perfectly
    # valid geometry, which is the worst possible failure: silent, plausible,
    # and it propagates into every downstream calculation.
    if len(body) != env.body_bytes:
        raise BodyMismatch(f"{env.body_key}: {len(body)} != {env.body_bytes}")
    if hashlib.sha256(body).hexdigest() != env.body_sha256:
        raise BodyMismatch(f"{env.body_key}: digest mismatch")

    return json.loads(body)

A truncated JSON document usually fails to parse, but a truncated feature collection frequently does not — it ends after a complete feature, and the reader gets a valid collection with fewer features than it should have. Nothing errors, and a coverage calculation quietly reports a smaller area.

Step 3 — Parse incrementally when the body is large

A hundred-megabyte feature collection loaded with json.loads costs roughly six hundred megabytes of Python objects. ijson yields one feature at a time and holds only that feature.

import ijson


async def stream_features(store, key: str):
    """Yield features one at a time, holding one feature's worth of memory.

    ijson drives the parse from the byte stream, so the peak footprint is
    the largest single feature rather than the whole document.
    """
    async with store.open(key) as raw:
        async for feature in ijson.items_async(raw, "features.item"):
            yield feature


async def total_area(store, key: str) -> float:
    total = 0.0
    async for feature in stream_features(store, key):
        total += shape(feature["geometry"]).area
    return total

The limit of the technique is that a single geometry larger than memory is still larger than memory: ijson can stream a collection of features, but the moment one feature is handed to shape() it is materialised whole. For genuinely enormous single geometries the answer is to simplify at the producer or to tile the feature, not to parse harder.

Step 4 — Chunk only when the parts stand alone

When the parts genuinely stand alone, the envelope needs a group identifier, a part index and a declared total, so a consumer can detect a missing part rather than waiting for it forever. The reassembly then belongs behind the same idempotency key the rest of the pipeline uses, as Event Key Generation for Spatial Data describes.

Spatial validation and error handling

Compute the bounding box before serialising, not after fetching. The whole point of the envelope is that a consumer can decide without the body. A bounding box derived at resolve time is derived too late.

Reject an envelope whose declared vertex count is absent. It is the cost estimate that backpressure charges against, and an event with no estimate is an event of unknown cost, which is the thing the budget exists to prevent.

Verify geometry validity after reassembly, never per chunk. A chunk of a multipolygon can be valid while the assembled whole has overlapping parts. Run the repair described in Geometry Validation Pipelines on the assembled geometry only.

Give object bodies a longer retention than the broker. If the topic retains seven days and the bucket expires objects after three, a replay on day four resolves references to objects that no longer exist — and the failure appears only during an incident, which is when replay is being used.

Retry, backoff and delivery guarantees

The claim check changes what a retry costs. The envelope is small, so redelivering it is nearly free; the body is content-addressed, so re-storing it is idempotent. A producer that crashes between the store write and the publish leaves an orphaned object and no event, which is a garbage-collection problem rather than a correctness one — sweep objects with no referencing event after a period longer than the broker’s retention.

The reverse order is a correctness problem, which is why the ordering in Step 1 matters. Publish-then-store means a consumer can resolve a reference before the body exists, producing a NoSuchKey that looks exactly like the object having been deleted, and retrying it burns the retry budget on a race that will resolve on its own.

Chunked events have the harder guarantee. At-least-once delivery means a part can arrive twice, so reassembly must be idempotent per part index; ordering is not guaranteed, so the reassembler cannot assume part n precedes part n+1; and a part can be lost entirely, so the buffer needs a timeout that fails the group rather than waiting. That is three failure modes the claim check simply does not have, and it is the strongest argument for preferring it.

Verification

import json
import pytest
from shapely.geometry import Polygon, MultiPolygon, mapping


@pytest.mark.asyncio
async def test_small_geometry_stays_inline(store, broker):
    """A back garden must not pay for an object-storage round trip."""
    small = mapping(Polygon([(0, 0), (0, 1), (1, 1), (1, 0)]))
    await publish(small, "f-1", "2026-08-08T10:00:00Z", store, broker)

    env = json.loads(broker.sent[-1])
    assert env["geometry"] is not None
    assert env["body_key"] is None
    assert store.writes == 0


@pytest.mark.asyncio
async def test_large_geometry_is_referenced_and_verifiable(store, broker):
    """The envelope must stay small and the digest must round-trip."""
    huge = mapping(MultiPolygon([
        Polygon([(i, j), (i, j + 0.9), (i + 0.9, j + 0.9), (i + 0.9, j)])
        for i in range(120) for j in range(120)
    ]))
    await publish(huge, "f-2", "2026-08-08T10:00:00Z", store, broker)

    env = json.loads(broker.sent[-1])
    assert env["geometry"] is None
    assert len(broker.sent[-1]) < 2048          # envelope, not payload
    assert await resolve(SpatialEnvelope(**env), store) == huge


@pytest.mark.asyncio
async def test_truncated_body_is_rejected(store, broker):
    """The failure this check exists for: a valid, smaller, wrong geometry."""
    huge = mapping(MultiPolygon([
        Polygon([(i, 0), (i, 1), (i + 0.9, 1), (i + 0.9, 0)]) for i in range(9000)
    ]))
    await publish(huge, "f-3", "2026-08-08T10:00:00Z", store, broker)
    env = SpatialEnvelope(**json.loads(broker.sent[-1]))

    store.truncate(env.body_key, keep=env.body_bytes // 2)
    with pytest.raises(BodyMismatch):
        await resolve(env, store)

The third test is the one worth keeping. Without the digest check it passes for the wrong reason on some inputs and fails silently on others, because whether a truncated GeoJSON document parses depends on where the cut fell.

Troubleshooting

FAQ

Related

• Spatial Payload Routing & Parsing — the section this topic belongs to
• Async Processing for Geometry-Heavy Payloads — the compute half of the same problem, once the geometry has arrived
• Protocol Buffers vs GeoJSON for High-Frequency Spatial Events — how much of the size problem a denser encoding removes
• Geometry Validation Pipelines — what to run on a geometry once it has been reassembled
• Backpressure & Flow Control for Spatial Consumers — where the envelope’s vertex count is charged against a work budget

Explore this section