Implementing async batch processing for large GIS datasets

You are ingesting a multi-county zoning fabric — roughly 1.8 million parcels spread across a dozen municipal shapefiles and GeoJSON overlays — and your synchronous script dies with MemoryError before the first spatial join even runs. This page answers one narrow question: how do you process a GIS dataset that is too large to fit in RAM, without blocking on slow municipal endpoints and without corrupting topology mid-run? The technique here is a concrete application of async batch processing, tuned for the moment a regional feed outgrows a single GeoDataFrame.read_file() call. The goal is a pipeline that streams fixed-size chunks, isolates network waits from geometry math, repairs invalid features in flight, and emits a per-chunk audit trail you can roll back to.

Diagnosis: identifying where a synchronous pipeline collapses

Before adding asyncio, confirm the failure is actually I/O- and memory-bound rather than algorithmic. Spatial ingestion failures rarely surface as clean exceptions; they appear as three distinct symptoms.

  1. Unbounded memory allocation. Loading every municipality into one frame triggers MemoryError: Unable to allocate 14.2 GiB for an array with shape (1840000, 128) and data type object. GeoPandas materializes the full dataset in RAM before it builds a spatial index, so peak memory scales with the largest county, not the median.
  2. Topology drift. Municipal cartographers publish multipart polygons with self-intersections and sliver geometries. Passed to sjoin, the GEOS backend raises TopologyException: found non-noded intersection between LINESTRING (...) and LINESTRING (...), aborting the run and leaving partial writes in the target table.
  3. Event-loop blocking. Synchronous requests.get() against a rate-limited portal, or a read off a slow NFS-mounted shapefile directory, parks the interpreter. The GIL means no geometry validation happens while the socket waits, so latency compounds into cascading timeouts.

Reproduce the memory profile before refactoring so you can measure the win:

import tracemalloc
import geopandas as gpd

tracemalloc.start()
gdf = gpd.read_file("county_parcels.gpkg", engine="pyogrio")
current, peak = tracemalloc.get_traced_memory()
print(f"current={current / 1e9:.2f} GB  peak={peak / 1e9:.2f} GB")
tracemalloc.stop()

If peak approaches or exceeds your container limit on a single file, chunking is the fix. A chunked async pass typically cuts peak RSS by 60–80% because only one window of features is resident at a time.

Decoupled async batch pipeline for large GIS datasets An async I/O lane polls municipal APIs and streams fixed-size windows into a chunk queue. A CPU-bound lane offloaded with asyncio.to_thread applies CRS transform and make_valid repair, then a validity decision routes features either to sjoin and commit or to a quarantine partition. ASYNC I/O LANE event loop · asyncio Poll municipal API fetch_with_backoff 429 → backoff + jitter Stream windows pyogrio · aiofiles yield 5k–10k rows socket waits never block the geometry math CHUNK QUEUE one window resident at a time CPU-BOUND LANE asyncio.to_thread · ProcessPoolExecutor CRS transform → EPSG:2263 make_valid repair topology fix in place valid? sjoin + commit PostGIS savepoint manifest · sha256 Quarantine failed_geometries .parquet yes no
Two decoupled lanes: the async I/O lane keeps the event loop free while a chunk queue meters work into the CPU-bound geometry lane, where a validity check commits clean features or quarantines the rest.

Architecture: decoupling I/O from CPU-bound geometry

The pipeline separates three execution contexts so a network wait never starves a spatial computation:

  • Async I/O layer. asyncio and aiofiles handle municipal API polling, rate limit management backoff, and streaming chunk reads without blocking the loop.
  • CPU-bound spatial layer. asyncio.to_thread (or a concurrent.futures.ProcessPoolExecutor for heavier joins) offloads CRS transformation, topology repair, and spatial joins, sidestepping GIL contention during the math.
  • Chunking window. Features are batched at 5,000–10,000 rows per iteration — large enough to amortize spatial-index rebuilds, small enough to keep each window’s frame well under the memory ceiling.
  • Per-chunk state. Each batch carries a deterministic manifest (source hash, CRS, feature count, validation status). Failed batches are quarantined, never silently dropped, which is what makes deterministic rollback and audit possible.

Step-by-step implementation

Each step below isolates exactly one concern. Compose them into the orchestrator at the end.

Step 1 — Size the chunk window

Pick a window that keeps a single frame’s geometry under your per-worker memory budget. For NY State Plane parcels with ~128 attribute columns, 5,000 rows holds peak working memory near 250–400 MB per chunk, leaving headroom for the spatial index. Make it a constant you can tune per deployment.

TARGET_CRS = "EPSG:2263"  # NY State Plane Long Island (feet)
CHUNK_SIZE = 5000
MAX_RETRIES = 3
BACKOFF_BASE = 2.0

Step 2 — Stream the source in fixed windows

Use pyogrio for fast vector reads and yield slices instead of returning the whole frame. The async generator lets the orchestrator await between chunks, so polling and disk reads can interleave.

import geopandas as gpd
from pathlib import Path
from typing import AsyncGenerator
import logging

logger = logging.getLogger("zoning_ingest")

async def stream_zoning_chunks(file_path: Path) -> AsyncGenerator[gpd.GeoDataFrame, None]:
    """Yield fixed-size chunks from a vector file using pyogrio for fast I/O."""
    try:
        full_gdf = gpd.read_file(file_path, engine="pyogrio")
        for i in range(0, len(full_gdf), CHUNK_SIZE):
            yield full_gdf.iloc[i:i + CHUNK_SIZE].copy()
    except Exception as exc:
        logger.error(f"Stream failure at {file_path}: {exc}")
        raise

For sources large enough that even the initial read_file is too heavy, swap to pyogrio.read_dataframe(path, skip_features=i, max_features=CHUNK_SIZE) so the windowing happens at the driver level and the full frame is never resident.

Step 3 — Enforce CRS alignment and repair topology per chunk

This is the only CPU-bound function, so keep it synchronous and pure — the orchestrator hands it to a worker thread. Detecting a missing CRS and projecting to the target is the same discipline covered in depth under CRS alignment strategies; here it runs inline on every window.

from shapely.validation import make_valid

def _normalize_and_validate(gdf_chunk: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
    """CRS alignment and topology repair for a single chunk."""
    if gdf_chunk.crs is None:
        logger.warning("Chunk missing CRS. Defaulting to EPSG:4326 before transform.")
        gdf_chunk = gdf_chunk.set_crs("EPSG:4326")

    gdf_chunk = gdf_chunk.to_crs(TARGET_CRS)

    # Repair self-intersections and ring-orientation defects in place
    invalid_mask = ~gdf_chunk.geometry.is_valid
    if invalid_mask.any():
        logger.info(f"Repairing {int(invalid_mask.sum())} invalid geometries in chunk.")
        gdf_chunk.loc[invalid_mask, "geometry"] = (
            gdf_chunk.loc[invalid_mask, "geometry"].apply(make_valid)
        )

    return gdf_chunk

Step 4 — Offload spatial work and build a per-chunk manifest

Wrap the CPU function with asyncio.to_thread so the event loop stays free, and capture a checksum + status the moment the chunk is validated. The manifest is what later feeds schema validation & data quality checks downstream.

import asyncio
import hashlib
from typing import Dict, Any

async def process_chunk(chunk: gpd.GeoDataFrame, chunk_idx: int) -> Dict[str, Any]:
    """Normalize on a worker thread, then emit a compliance manifest."""
    try:
        pre_invalid = int((~chunk.geometry.is_valid).sum())
        normalized = await asyncio.to_thread(_normalize_and_validate, chunk)

        chunk_hash = hashlib.sha256(normalized.to_json().encode()).hexdigest()[:16]
        manifest = {
            "chunk_id": chunk_idx,
            "feature_count": len(normalized),
            "crs": normalized.crs.to_string(),
            "invalid_repaired": pre_invalid,
            "checksum": chunk_hash,
            "status": "validated",
        }
        return {"gdf": normalized, "manifest": manifest}
    except Exception as exc:
        logger.error(f"Chunk {chunk_idx} processing failed: {exc}")
        return {"gdf": None, "manifest": {"chunk_id": chunk_idx, "status": "failed", "error": str(exc)}}

Step 5 — Add bounded retries for the network-bound fetch

When the source is an OGC API rather than a local file, every fetch needs exponential backoff with jitter so a 429 storm does not synchronize your retries into a thundering herd. This mirrors the broader error handling & retry logic patterns used across the ingestion layer.

import random

async def fetch_with_backoff(session, url: str, retries: int = MAX_RETRIES):
    for attempt in range(retries):
        try:
            async with session.get(url) as resp:
                if resp.status == 429:
                    wait = BACKOFF_BASE ** attempt + random.uniform(0.0, 1.0)
                    await asyncio.sleep(wait)
                    continue
                resp.raise_for_status()
                return await resp.json()
        except Exception:
            if attempt == retries - 1:
                raise
            await asyncio.sleep(BACKOFF_BASE ** attempt)

Step 6 — Orchestrate the run and persist the audit trail

The orchestrator drives the generator, accumulates manifests, quarantines failures, and writes the manifest with aiofiles so the final write does not block the loop either.

import aiofiles
import json
from typing import List

async def run_ingestion_pipeline(source_file: Path):
    """Main async pipeline orchestrator."""
    compliance_artifacts: List[Dict[str, Any]] = []
    failed_chunks: List[int] = []

    logger.info(f"Initializing async batch processing for: {source_file.name}")

    async for chunk in stream_zoning_chunks(source_file):
        chunk_idx = len(compliance_artifacts)
        result = await process_chunk(chunk, chunk_idx)
        compliance_artifacts.append(result["manifest"])

        if result["gdf"] is not None:
            logger.info(f"Chunk {chunk_idx} ready for spatial join and commit.")
        else:
            failed_chunks.append(chunk_idx)

    manifest_path = source_file.parent / f"{source_file.stem}_compliance_manifest.json"
    async with aiofiles.open(manifest_path, "w") as f:
        await f.write(json.dumps(compliance_artifacts, indent=2))

    logger.info(
        f"Pipeline complete. {len(compliance_artifacts)} chunks processed. "
        f"{len(failed_chunks)} quarantined."
    )
    return compliance_artifacts

Verification & testing

Confirm the refactor actually fixed the three failure modes rather than hiding them.

  • Memory ceiling held. Re-run the tracemalloc probe wrapped around run_ingestion_pipeline. Peak should now track CHUNK_SIZE, not total feature count — expect it flat across a 50k-feature file and a 1.8M-feature file.
  • Feature conservation. Assert no rows vanished: sum(m["feature_count"] for m in artifacts if m["status"] == "validated") + quarantined == source_total. A mismatch means a chunk silently failed before its manifest was appended.
  • Spatial validity. After normalization, every committed geometry must pass gdf.geometry.is_valid.all(). Spot-check repaired features with make_valid did not collapse a polygon to a GeometryCollection you cannot store.
  • Deterministic checksums. Re-running the same source must reproduce identical per-chunk checksum values. Drift here points to non-deterministic ordering — pin the read order before chunking.
  • Log signature. A healthy run logs one Repairing N invalid geometries line only for chunks that need it and a final Pipeline complete. X chunks processed. 0 quarantined. Any non-zero quarantine count is your signal to inspect the manifest.

A reference comparison for tuning the window:

Chunk size Peak RSS / chunk Index rebuild overhead Commit latency Best for
1,000 ~80 MB high (frequent rebuilds) low tight memory containers
5,000 ~300 MB balanced balanced most regional feeds
10,000 ~600 MB low higher per-commit wide containers, few joins

Failure recovery

When a chunk fails mid-pipeline, the run must continue and stay reproducible.

  • Quarantine, do not abort. A TopologyException or GEOSException on one window writes a status: failed manifest entry and routes the offending features to a failed_geometries.parquet partition with their source coordinates and the error trace. Use shapely.validation.explain_validity() to record the exact defect for targeted reprocessing once the municipality corrects its data.
  • Resume from the last boundary. Persist the last committed chunk_id to a Redis-backed offset store. On restart, skip windows already present in the manifest so reruns are idempotent and do not double-write to the target table.
  • Circuit-break on pressure. If a chunk exceeds ~120 seconds or RSS crosses ~80% of the container limit, flush pending transactions, write the current offset, and emit a PIPELINE_PAUSED event. Restore to the last committed chunk boundary with a transactional savepoint or WAL replay.
  • Reprocess in isolation. Quarantined parcels feed a narrow re-run against only the failed chunk_ids, which is cheap because the manifest already records exactly which windows to retry.

Frequently asked questions

Why use asyncio.to_thread instead of running the spatial join in the event loop?

Geometry operations are CPU-bound and hold the GIL, so running them directly on the loop would block every concurrent fetch and read. Offloading to a worker thread keeps the loop free for I/O. For very heavy joins across many cores, move to a ProcessPoolExecutor instead, which sidesteps the GIL entirely at the cost of pickling each chunk.

How do I pick the chunk size for my dataset?

Start at 5,000 rows and measure peak RSS per chunk with tracemalloc. Halve it if you approach the container limit; double it if rebuild overhead dominates and you have memory headroom. Attribute width matters as much as row count — a 200-column parcel table needs a smaller window than a slim geometry-only feed.

What happens to features that make_valid cannot repair cleanly?

make_valid can return a GeometryCollection mixing polygons, lines, and points when it splits a badly self-intersecting shape. If your target column only accepts polygons, filter the collection to its polygonal parts before commit and route the rest to the failed_geometries.parquet quarantine for manual review rather than forcing an invalid write.

Can this pipeline write straight to PostGIS, or should it stage first?

Commit each validated chunk inside a transaction with a savepoint at the chunk boundary so a later failure rolls back only the open window, not the whole run. Staging to GeoParquet first is worthwhile when the same normalized output also feeds GIS export sync workflows, since you avoid re-reading PostGIS to regenerate exports.