Async Execution for Large Datasets

10 min read

Async execution is the throughput pattern within Spatial Test Pattern Design & Implementation that lets a validation suite keep up with enterprise-scale geospatial data without sacrificing deterministic outcomes. Modern GIS QA pipelines routinely ingest multi-terabyte vector mosaics, LiDAR point clouds, and raster time series where a synchronous, feature-by-feature loop introduces unacceptable latency, memory exhaustion, and CI/CD timeout failures. This pattern reshapes validation from a monolithic blocking process into a streaming architecture built on non-blocking I/O, bounded concurrency, and strict memory-safe execution boundaries — so that the same strict tolerance thresholds you enforce on a thousand-feature fixture still hold when you validate a billion features across a worker pool.

The core idea is that concurrency must never change the verdict. An async pipeline that produces different pass/fail results depending on scheduling order is not a faster validator — it is a non-deterministic one. Everything below is organized around preserving exact, reproducible verdicts while scaling horizontally.

The runtime shape that carries this is a streaming producer/consumer pipeline: a chunker partitions the dataset, a bounded queue paces the work, a fixed pool of workers validates each chunk, and a result sink aggregates verdicts while feeding memory pressure back to the chunker.

Execution strategy taxonomy

Choosing a concurrency model is a function of what each validation rule actually does to a chunk. I/O-bound checks (asset retrieval, format parity reads) starve under threads but thrive on coroutines; CPU-bound checks (topology, reprojection) need real parallelism via processes. The table below maps the common spatial workloads to a recommended model, the unit each chunk should be measured in, and a starting bound for concurrency.

The recurring split is I/O-bound versus CPU-bound. A single orchestrator can host both: an asyncio event loop dispatches I/O coroutines directly and pushes CPU-bound work onto a process pool via loop.run_in_executor, keeping the loop responsive while heavy geometry math runs on dedicated cores.

Pipeline-first architecture and memory-safe execution

Memory safety is non-negotiable when processing large spatial datasets. Loading an entire GeoParquet, Shapefile, or GPKG layer into RAM before validation guarantees an OOM kill in a containerized CI runner. The pattern enforces a streaming, chunk-based ingestion strategy: data is partitioned by spatial index (H3, S2, or quadtree tiles) or by feature count, then dispatched to worker coroutines through a bounded queue.

# async_validation_config.yaml
execution:
  mode: async
  chunk_strategy: spatial_index
  chunk_size_mb: 256
  max_concurrent_workers: 8
  backpressure_threshold: 0.75
  memory_limit_mb: 4096
  timeout_per_chunk_s: 120
  retry_policy:
    max_attempts: 3
    exponential_backoff: true

The design decouples ingestion, validation execution, and result aggregation. Workers pull chunks from a bounded queue, apply validation rules, and push structured results — pass/fail, error geometries, metric deltas — to a centralized sink. The bound on the queue is what makes the pipeline memory-safe: it caps the number of in-flight chunks so peak resident memory is roughly max_concurrent_workers × chunk_size_mb plus per-worker overhead, regardless of total dataset size.

Backpressure closes the loop. When the sink observes that resident memory has crossed backpressure_threshold of memory_limit_mb, it stops the chunker from enqueuing new work until in-flight chunks drain. In asyncio, an asyncio.Queue(maxsize=N) provides this for free: await queue.put(chunk) suspends the producer when the queue is full, so the chunker naturally paces itself to consumer throughput without a manual polling loop.

Bounded concurrency and the orchestrator

The orchestrator owns the event loop and the lifecycle of the worker set. Bounded concurrency means there is always an explicit ceiling on simultaneous work — never an unbounded asyncio.gather over a generator of a billion features. A Semaphore (for coroutine fan-out) or a fixed-size queue with a fixed worker count (for the producer/consumer shape) both express the bound; the queue shape is preferred for validation because it gives you backpressure and graceful drain in one structure.

# inline sketch: a Semaphore caps coroutine fan-out
sem = asyncio.Semaphore(max_concurrent_workers)

async def guarded(chunk):
    async with sem:            # never more than N validations at once
        return await validate_chunk(chunk)

Strict tolerance configuration and state management

Spatial QA requires deterministic tolerance enforcement across distributed workers, and async execution introduces a real state-propagation hazard: tolerance thresholds, CRS definitions, and precision rules must be serialized into every worker context without drift. A worker that inherits a slightly different xy_tolerance — or silently falls back to a process-default CRS — will produce verdicts that disagree with its siblings.

The pattern is to freeze configuration into an immutable object at pipeline bootstrap, validate it once, and inject the same frozen instance into each worker. Tolerance configuration must explicitly define:

• Geometric precision — coordinate rounding and vertex snapping thresholds applied before any spatial predicate is evaluated.
• CRS alignment — mandatory on-the-fly reprojection or a strict CRS-match gate that fails fast rather than distorting metrics silently.
• Attribute drift limits — acceptable variance for numeric fields, string normalization rules, and null-handling policy.

For numeric comparisons the verdict should rest on a relative error bound rather than a raw difference, so the threshold scales with coordinate magnitude. A measured value $\hat{x}$ passes against expected $x$ when

where $\tau$ is the configured relative tolerance and $\varepsilon_0$ guards the division near zero. Freezing $\tau$ and $\varepsilon_0$ at bootstrap is what eliminates the floating-point accumulation and per-worker environment divergence that otherwise make concurrent runs non-reproducible. This directly supports geometry validation patterns: ring-closure checks, self-intersection detection, and minimum-area thresholds are then evaluated against identical mathematical baselines on every core.

Domain-specific worker routing

Async execution does not replace spatial validation logic; it accelerates its application by routing each validation domain to the pool best suited to its computational profile.

• Topology rule enforcement — CPU-intensive operations such as adjacency validation, sliver-polygon detection, and network connectivity checks are offloaded to isolated ProcessPoolExecutor instances, then aggregated into a single violation report keyed by feature ID and rule code. See topology rule enforcement for the rule set itself.
• Attribute and metadata checks — lightweight schema validation, enum compliance, and metadata-completeness checks run on high-throughput I/O coroutines, parallelizing across thousands of features per second without saturating CPU cores. The check definitions live in attribute and metadata checks.
• Cross-format parity testing — async workers read source and target formats (for example GeoJSON to GeoParquet) concurrently, compute geometry hashes, and compare attribute deltas so that disk or network latency never stalls the parity loop. The comparison contract is defined in cross-format parity testing.

Production-grade Python implementation

The orchestrator below partitions a GeoParquet dataset by row-group, validates each chunk inside a process pool with frozen tolerances, and asserts an aggregate verdict from pytest. It targets Shapely 2.x, GeoPandas 0.14+, and pytest 7+.

# test_async_validation.py
import asyncio
import os
from dataclasses import dataclass
from concurrent.futures import ProcessPoolExecutor

import geopandas as gpd
import pyarrow.parquet as pq
import pytest
from shapely import make_valid  # Shapely 2.x
from shapely.validation import explain_validity


@dataclass(frozen=True)
class Tolerances:
    xy_tolerance: float = 1e-6      # ~0.1 m near the equator in EPSG:4326
    min_area: float = 1e-9
    target_crs: str = "EPSG:4326"


def validate_chunk(parquet_path: str, row_group: int, tol: Tolerances) -> dict:
    """Runs in a worker process: pure, deterministic, no shared mutable state."""
    table = pq.ParquetFile(parquet_path).read_row_group(row_group)
    gdf = gpd.GeoDataFrame.from_arrow(table)

    # Strict CRS gate — fail fast rather than silently reproject.
    if gdf.crs is None or gdf.crs.to_string() != tol.target_crs:
        return {"row_group": row_group, "failures": ["crs_mismatch"], "count": len(gdf)}

    failures = []
    for fid, geom in zip(gdf.index, gdf.geometry, strict=True):
        if geom is None or geom.is_empty:
            failures.append((fid, "empty_geometry"))
            continue
        if not geom.is_valid:
            failures.append((fid, explain_validity(geom)))
            continue
        if geom.area < tol.min_area and geom.geom_type in ("Polygon", "MultiPolygon"):
            failures.append((fid, "below_min_area"))
    return {"row_group": row_group, "failures": failures, "count": len(gdf)}


async def run_validation(parquet_path: str, tol: Tolerances, max_workers: int) -> dict:
    loop = asyncio.get_running_loop()
    n_groups = pq.ParquetFile(parquet_path).num_row_groups
    sem = asyncio.Semaphore(max_workers)  # bound in-flight chunks

    with ProcessPoolExecutor(max_workers=max_workers) as pool:
        async def one(rg: int) -> dict:
            async with sem:  # backpressure: never exceed max_workers chunks
                return await loop.run_in_executor(pool, validate_chunk, parquet_path, rg, tol)

        results = await asyncio.gather(*(one(rg) for rg in range(n_groups)))

    total = sum(r["count"] for r in results)
    failures = [f for r in results for f in r["failures"]]
    return {"features": total, "failures": failures}


@pytest.mark.parametrize("dataset", ["fixtures/parcels_50m.parquet"])
def test_large_dataset_is_valid(dataset):
    tol = Tolerances()  # frozen config injected identically into every worker
    report = asyncio.run(run_validation(dataset, tol, max_workers=os.cpu_count()))
    assert report["failures"] == [], (
        f"{len(report['failures'])} invalid features across {report['features']}"
    )

Two properties make this deterministic: validate_chunk is a pure function of (path, row_group, tol) with no shared mutable state, and the frozen Tolerances dataclass is pickled identically into every process. Reordering chunks or changing max_workers cannot change the failure set — only how fast it is produced.

PostGIS / database-side counterparts

When the dataset already lives in PostGIS, push the heavy geometry work to the database and let the async layer parallelize over spatial partitions rather than over rows in Python. Each coroutine validates one tile of a spatial grid, and ST_SubdivideHints-style tiling keeps per-query memory bounded.

-- Validate one tile; the async layer fans this out per grid cell.
SELECT feature_id, ST_IsValidReason(geom) AS reason
FROM   parcels
WHERE  geom && ST_TileEnvelope(:zoom, :x, :y)   -- GiST-indexed bbox prefilter
  AND  NOT ST_IsValid(geom);

# Async fan-out over tiles with asyncpg; bound concurrency with a pool size.
import asyncpg

async def validate_tiles(dsn, tiles, pool_size=8):
    pool = await asyncpg.create_pool(dsn, min_size=pool_size, max_size=pool_size)
    sem = asyncio.Semaphore(pool_size)

    async def one(z, x, y):
        async with sem, pool.acquire() as conn:
            return await conn.fetch(
                "SELECT feature_id, ST_IsValidReason(geom) AS reason "
                "FROM parcels WHERE geom && ST_TileEnvelope($1,$2,$3) "
                "AND NOT ST_IsValid(geom)", z, x, y)

    rows = await asyncio.gather(*(one(z, x, y) for (z, x, y) in tiles))
    return [r for batch in rows for r in batch]

Bounding the connection pool to the same number as the semaphore prevents the classic failure where async fan-out opens thousands of connections and exhausts max_connections on the server.

Pipeline integration and observability

Platform and DevOps teams should treat async spatial validation as a first-class CI/CD stage, governed by the same precision rules as the rest of the core geospatial QA architecture. Three controls matter most.

• Resource guardrails — set memory_limit_mb and a CPU quota in the Kubernetes or Compose manifest, and pin native dependency versions (libgeos, PROJ, GDAL) in the runner image so floating-point behavior is identical between local and CI. A SIGTERM-aware drain routine should finish in-flight chunks before the process exits, so a graceful shutdown never reports false failures.
• Structured telemetry — instrument workers with OpenTelemetry or Prometheus and track chunks_processed, validation_failures, backpressure_events, and worker_pool_utilization, correlated by a trace ID so a slow spatial predicate can be isolated to a specific tile. Emit failures as machine-readable records (feature ID, rule code, reason) rather than log strings.
• Retry and circuit breaking — apply exponential backoff to transient I/O failures (object-store throttling, network timeouts) and a circuit breaker to fail fast when a downstream service degrades, preserving runner capacity instead of burning the CI budget on doomed retries.

For fixtures large enough to exercise this path, generate them with the test data generation and mocking strategies patterns rather than copying production extracts into the repo.

Common failure modes and gotchas

1. Unbounded fan-out. asyncio.gather over a generator of every feature schedules them all at once; resident memory tracks the dataset, not the worker bound. Always gate fan-out with a Semaphore or a maxsize queue.
2. Tolerance drift between workers. A worker that re-reads config from the environment, or inherits a process-default CRS, produces verdicts that disagree with its siblings. Freeze tolerances at bootstrap and pickle one immutable instance into every process.
3. Chunk boundaries splitting features. Partitioning by byte offset can cut a multi-part feature across two chunks, hiding a topology violation that only appears when the parts are evaluated together. Partition by spatial index or complete feature, never mid-record.
4. Connection-pool exhaustion. PostGIS fan-out that acquires a connection per coroutine without a bounded pool will exceed max_connections and fail under load. Bind pool size to the concurrency semaphore.
5. Anti-meridian and polar tiles. Spatial-index chunking near ±180° longitude or the poles produces envelopes that wrap or degenerate; a tile prefilter can then silently drop features. Test the chunker explicitly against anti-meridian and polar fixtures.
6. Lost exceptions in detached tasks. A coroutine launched with asyncio.create_task and never awaited can fail silently, so a chunk is skipped and the suite reports green. Await every task (or use asyncio.TaskGroup on Python 3.11+, which propagates the first exception).
7. CPU work on the event loop. Running Shapely topology checks directly inside a coroutine blocks the loop and serializes everything; offload CPU-bound predicates to run_in_executor with a process pool.

Conclusion

Async execution turns geospatial QA from a sequential bottleneck into a horizontally scalable stage without ever changing a verdict: memory-safe chunking caps peak RAM, bounded concurrency caps in-flight work, and frozen tolerances keep every worker mathematically identical. Apply it as the throughput layer of Spatial Test Pattern Design & Implementation, and the same deterministic checks that guard a small fixture will guard a petabyte-scale dataset with predictable latency and audit-ready traceability.

Related

• Geometry Validation Patterns
• Topology Rule Enforcement
• Attribute & Metadata Checks
• Cross-Format Parity Testing
• Setting Up Spatial Tolerance Thresholds in Assertions
• Up: Spatial Test Pattern Design & Implementation