Streaming GeoJSON with ijson in Async Consumers

Drive the parse from the byte stream with ijson so peak memory is the largest single feature rather than the whole document, run it off the event loop because the parsing work is synchronous CPU, and check completeness explicitly — a truncated document streams every feature it does have before raising.

This guide sits under Streaming & Chunking Large Geometry Payloads, within Spatial Payload Routing & Parsing. It solves the consumer’s memory limit; the broker’s size limit is solved by The Claim-Check Pattern for Oversized Spatial Payloads, and a pipeline handling large geometry usually needs both.

When to use this pattern

  • Documents are large enough that resident memory is a constraint — tens of megabytes upward for coordinate-dense data.
  • The work per feature is independent, so features can be processed and released rather than collected.
  • The consumer is async, which makes the blocking-parse problem real rather than theoretical.

Where the memory goes

The whole document, or one feature Resident memory is traced through the processing of a hundred-megabyte FeatureCollection. With json.loads, memory climbs steadily during the parse to roughly six hundred megabytes and stays there for the whole processing pass, because every feature is alive until the document object is released — a coordinate pair that occupies about twenty bytes of JSON text becomes a list object plus two float objects, well over a hundred bytes of heap, so coordinate-dense data expands several-fold. With ijson the profile is flat at a few megabytes with small spikes, each spike being one feature materialised, processed and released before the next is read. The peak is therefore set by the largest single feature rather than by the document, which matters because the largest feature is a property of the data and the document size is a property of how the data was batched. The limit of the technique is visible in the same shape: a single feature larger than memory is still larger than memory, because the moment it is handed to shape() it is materialised whole. resident memory · one 100 MB FeatureCollection 600 MB 0 json.loads — every feature alive until the document object is released ijson — one feature materialised, processed, released; peak is the largest feature The limit of the technique is visible in the same shape: a single feature larger than memory is still larger than memory, because the moment it reaches shape() it is materialised whole. For those, simplify or tile at the producer — do not parse harder.
Figure 1. The peak becomes a property of the data rather than of how the data was batched, which is what makes a consumer's memory limit predictable.

Complete runnable implementation

python 
import asyncio
import hashlib
from dataclasses import dataclass

import ijson
from shapely.geometry import shape


class TruncatedDocument(Exception):
    """The stream ended before the document did, or before its declared count."""


@dataclass(slots=True)
class StreamResult:
    features_seen: int
    total_area: float


def _parse_blocking(fileobj, declared_count: int | None, handle) -> StreamResult:
    """Run the parse synchronously; the caller keeps it off the loop.

    ijson's parsing work is CPU on the calling thread, and on coordinate-dense
    data that is enough to stall an event loop for hundreds of milliseconds —
    which surfaces as consumer heartbeat failures, not as slow parsing.
    """
    seen, total = 0, 0.0
    try:
        for feature in ijson.items(fileobj, "features.item"):
            handle(feature)
            total += shape(feature["geometry"]).area
            seen += 1
            # `feature` is released here. Appending it to a list — even to
            # "collect results" — reinstates the whole memory problem.
    except ijson.IncompleteJSONError as exc:
        raise TruncatedDocument(f"stream ended after {seen} features") from exc

    # A cut that landed after a complete feature parses cleanly to that point.
    # Only a declared count catches it.
    if declared_count is not None and seen != declared_count:
        raise TruncatedDocument(f"expected {declared_count} features, saw {seen}")

    return StreamResult(features_seen=seen, total_area=total)


async def stream_features(fileobj, declared_count: int | None = None,
                          handle=lambda f: None) -> StreamResult:
    """Parse off the event loop, so the consumer keeps polling."""
    return await asyncio.to_thread(_parse_blocking, fileobj, declared_count, handle)


async def verify_then_stream(store, key: str, expected_sha256: str,
                             declared_count: int) -> StreamResult:
    """Verify the whole body first when correctness matters more than memory.

    Digesting requires reading every byte, but not holding them: read in
    fixed-size blocks, hash them, discard them. Peak memory is the block.
    """
    digest = hashlib.sha256()
    async for block in store.iter_blocks(key, size=1 << 20):
        digest.update(block)
    if digest.hexdigest() != expected_sha256:
        raise TruncatedDocument(f"{key}: digest mismatch before parsing")

    async with store.open(key) as fileobj:
        return await stream_features(fileobj, declared_count)

Verifying before parsing costs a second pass over the bytes and removes the entire class of “processed half a document and committed it” failures. Where a second pass is unaffordable, the declared feature count is the cheaper approximation.

The features it does have are processed before it fails A document declaring twelve thousand features is truncated at seventy per cent, with the cut falling immediately after a complete feature. Incremental parsing yields eight thousand four hundred features, every one of them valid, and the consumer processes each as it arrives — writing to a database, emitting downstream events, updating an area total. Only at the end does the parser reach the truncation and raise. By then the work is done and, in a consumer that commits per feature, committed: the pipeline has recorded a coverage figure that is seventy per cent of the truth with no indication that anything is missing. A whole-document parse fails before any work happens, which is worse for latency and much better for correctness. The fix is not to abandon streaming but to make completeness explicit — a declared feature count checked at the end, or a digest verified over the whole body before parsing begins — so the consumer knows whether the eight thousand four hundred features it processed were all of them. document declares 12 000 features · truncated at 70%, after a complete feature 8 400 features — every one valid, every one processed 3 600 features that are not there the cut streaming, without a completeness check work is done — and, per-feature commits, committed a coverage figure that is 70% of the truth, with nothing to indicate anything is missing with a declared count or a pre-verified digest count checked at the end: 8 400 ≠ 12 000 → raise digest verified first: nothing is processed at all the second costs a pass over the bytes and is worth it A whole-document parse fails before any work happens — worse for latency, much better for correctness. Streaming trades that away and has to buy it back.
Figure 2. Incremental parsing moves the failure from before the work to after it. That is the cost of the memory saving, and it is payable.

Parameter reference

Name Type Spatial constraint Default
Prefix str "features.item" for a FeatureCollection; "geometries.item" for a GeometryCollection —
declared_count int | None From the envelope; the cheap completeness check None
Block size int For digesting without holding the body 1 MiB
asyncio.to_thread call Required — the parse is synchronous CPU on the calling thread —
Backend str ijson.backends.yajl2_c is far faster than the pure-Python one auto
Accumulation — None. Appending features reinstates the whole problem —

Gotchas and spatial edge cases

  1. ijson.items returns coordinates as Decimal by default in some backends. shape() accepts them, but arithmetic mixing Decimal and float raises, and the memory advantage shrinks because Decimal objects are larger than floats. Use ijson.items(..., use_float=True) where the backend supports it, and assert the type in a test rather than discovering it in production.

  2. A single feature can still exhaust memory. Streaming bounds the document, not the feature: one national coastline handed to shape() is materialised whole. For those the answer is at the producer — simplify for transport, or tile the feature — not a different parser.

  3. The blocking parse stalls the event loop, and the symptom is a rebalance. A consumer that stops polling for four hundred milliseconds misses heartbeats, gets evicted, and its partitions are reassigned; the batch is then redelivered to another consumer which does the same. It presents as a rebalance storm rather than as slow parsing, which sends people to look at broker configuration.

  4. Prefixes are silently wrong rather than erroneous. A typo in "features.item" yields nothing at all — no exception, no warning, just a document with zero features. Assert a non-zero count, or a schema change that renames the array produces a clean, empty, wrong result.

  5. Streaming and the claim check solve different limits. The document still had to arrive, so streaming does nothing about a broker’s per-message ceiling. A pipeline handling gigabyte-scale geometry needs the claim check to move it and ijson to read it.

  6. Per-feature commits make truncation unrecoverable. If the consumer commits its offset as it goes, a truncated document leaves the pipeline believing it processed the whole thing. Commit once at the end of a document, or verify the digest before starting.

A blocking parse presents as a rebalance storm, not as slow parsing A consumer parses a large collection synchronously on its event loop. The parse occupies the loop for four hundred milliseconds at a stretch, during which the heartbeat coroutine cannot run: it is scheduled, ready, and never given the loop. The broker sees missed heartbeats, presumes the member dead and rebalances the group, so the partitions move and the batch is redelivered to another member — which parses it the same way and meets the same fate. What appears in the incident channel is a rebalance storm, which sends the investigation to the broker configuration, to session timeouts, and to the network, none of which are the cause. Running the same parse in a thread leaves the loop free to service the heartbeat between awaits, so the member stays in the group and the only visible effect is that the parse takes as long as it takes. The symptom and the cause are in completely different subsystems, which is why the thread offload is worth a comment rather than being left as an idiom. parse on the event loop ijson parse — 400 ms, uninterrupted heartbeats scheduled, ready, never given the loop broker presumes the member dead → rebalance → batch redelivered → the next member does the same the incident channel shows a rebalance storm, and the investigation goes to session timeouts and the network parse in a thread same parse, off the loop heartbeats run · the member stays in the group Symptom and cause sit in different subsystems, which is why the thread offload deserves a comment rather than being left as an idiom.
Figure 3. Nothing about the symptom points at the parser, so this is a failure that is diagnosed by knowing the mechanism rather than by reading the logs.

Verification

python 
import io
import json
import pytest
from shapely.geometry import Polygon, mapping


def collection(n: int) -> bytes:
    return json.dumps({
        "type": "FeatureCollection",
        "features": [
            {"type": "Feature", "properties": {"i": i},
             "geometry": mapping(Polygon([(i, 0), (i, 1), (i + 1, 1), (i + 1, 0)]))}
            for i in range(n)
        ],
    }).encode()


@pytest.mark.asyncio
async def test_streams_every_feature():
    result = await stream_features(io.BytesIO(collection(5000)), declared_count=5000)
    assert result.features_seen == 5000


@pytest.mark.asyncio
async def test_truncation_after_a_complete_feature_is_caught():
    """The failure incremental parsing makes easy to miss."""
    raw = collection(5000)
    cut = raw[: raw.index(b'{"type": "Feature", "properties": {"i": 3500}')]
    with pytest.raises(TruncatedDocument):
        await stream_features(io.BytesIO(cut + b"]}"), declared_count=5000)


@pytest.mark.asyncio
async def test_wrong_prefix_produces_zero_rather_than_an_error():
    """Documented, and asserted, because it is silent."""
    def parse(fileobj):
        return sum(1 for _ in ijson.items(fileobj, "featurez.item"))

    assert parse(io.BytesIO(collection(100))) == 0


@pytest.mark.asyncio
async def test_parse_does_not_block_the_event_loop():
    """A stalled loop shows up as a rebalance, not as slow parsing."""
    ticks = 0

    async def heartbeat():
        nonlocal ticks
        while True:
            ticks += 1
            await asyncio.sleep(0.01)

    beat = asyncio.create_task(heartbeat())
    await stream_features(io.BytesIO(collection(40_000)))
    beat.cancel()
    assert ticks > 5, "the event loop was starved during the parse"

The second test constructs the truncation deliberately at a feature boundary, because a cut in the middle of a coordinate array raises on its own and proves nothing — the dangerous cut is the tidy one.