#Building Batch Conversion Pipelines with Python

Modern geospatial infrastructure demands columnar, cloud-optimized storage formats. Legacy vector datasets — particularly Shapefiles, File Geodatabases, and early GeoJSON implementations — carry rigid schemas, poor compression, and inefficient I/O patterns that compound painfully at scale. Building a reliable batch conversion pipeline in Python requires a disciplined approach that balances throughput, memory constraints, and strict metadata fidelity across thousands of files.

This guide covers a production-tested architecture for migrating large vector dataset libraries to GeoParquet, integrating schema normalization, parallel execution, geometry validation, and structured error routing. It is part of the Data Conversion & Migration Pipelines framework, which emphasises idempotent workflows, explicit error routing, and strict version control for spatial libraries.

#Pipeline architecture

The diagram below shows the six-stage flow of a production batch pipeline, from source inventory through to validated output and dead-letter routing for failed files.

#Prerequisites

Before implementing the pipeline, confirm your environment meets these requirements:

• Python 3.10+ with strict virtual environment isolation (venv or conda)
• Core libraries: geopandas>=1.0, pyarrow>=15.0, shapely>=2.0, pyogrio>=0.7, pandas>=2.2
• Cloud SDKs: boto3 (AWS S3), gcsfs (Google Cloud), or adlfs (Azure Data Lake)
• System dependencies: GDAL 3.6+, PROJ 9+, and libspatialindex for any spatial indexing steps
• Infrastructure: minimum 16 GB RAM per parallel worker node, NVMe-backed scratch storage, and a task orchestrator (Prefect, Airflow, or AWS Step Functions) for production deployments
• Monitoring: structured logging sink (CloudWatch, GCP Logging, or a local logging.FileHandler) and a dead-letter store (S3 prefix, GCS bucket, or a local quarantine directory)

Install with explicit version pins to prevent silent ABI breaks between C-extensions:

pip install "geopandas==1.0.1" "pyarrow==15.0.0" "shapely==2.0.4" "pyogrio==0.7.2" "pandas==2.2.1"

#Architectural foundations

The pipeline rests on three properties of the GeoParquet/Arrow stack that legacy formats cannot provide:

Columnar predicate push-down. Arrow organises data by column, so reading only the geometry column to verify topology — without pulling attribute columns into RAM — is a first-class operation. The entire probe_dataset step below exploits this.

Strict schema enforcement. Unlike Shapefile’s loose type system (all attributes are strings or doubles), Arrow demands explicit typed arrays. This makes schema normalization a hard prerequisite, not an afterthought, and catches type drift that would silently corrupt Shapefile-based pipelines.

Process-level isolation. GDAL maintains global driver state. Running each file conversion in a separate ProcessPoolExecutor worker (see step 3) prevents CRS registry corruption and GDAL open-file limits from cascading across the batch. This is why thread-based concurrency fails at scale for this workload.

For background on the Shapefile limitations that motivate this migration, and the ZSTD compression levels that govern output size, refer to those dedicated guides.

#Step-by-step workflow

#Step 1 — Inventory & lightweight probing

Scanning source directories or cloud buckets to catalogue input files must be decoupled from heavy I/O. Extract metadata — CRS, feature count, bounding box, and geometry type — without loading full datasets into memory. Use pyogrio.read_info() for lightweight probing.

# pyogrio>=0.7.2  geopandas>=1.0
import pyogrio
from pathlib import Path


def probe_dataset(filepath: Path) -> dict:
    """Extract per-file metadata without loading geometries into RAM.

    Returns a dict suitable for workload partitioning and CRS audit logs.
    """
    try:
        meta = pyogrio.read_info(str(filepath))
        return {
            "filepath": str(filepath),
            "crs": meta.get("crs"),
            "geometry_type": meta.get("geometry_type"),
            "feature_count": meta.get("features"),
            "bbox": meta.get("bounds"),
            "error": None,
        }
    except Exception as exc:
        return {"filepath": str(filepath), "error": str(exc)}

This step enables intelligent routing. Datasets exceeding memory thresholds or carrying invalid CRS definitions can be flagged for specialist handling before they consume worker resources. Probing also lets you partition workloads by estimated file size rather than distributing files blindly.

#Step 2 — Schema normalization & type casting

Legacy formats frequently contain mixed-type columns, implicit null geometries, or inconsistent date representations. Arrow enforces strict typing, making implicit coercion dangerous. Apply explicit type casting before serialization. For a full treatment of coercion strategies including nested JSON fields and multi-geometry types, see Schema Mapping for Legacy to Modern Formats.

Key operations:

• Cast object columns to explicit string or int64 types
• Standardise datetime columns to UTC-naive timestamp[ms] (Arrow’s canonical form)
• Flatten nested dictionaries into prefixed columns
• Harmonise None vs NaN in numeric fields

# pandas>=2.2  geopandas>=1.0
import pandas as pd
import geopandas as gpd


def normalize_schema(gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
    """Enforce Arrow-compatible types on all non-geometry columns.

    Mutates a copy; the original GeoDataFrame is not modified.
    """
    gdf = gdf.copy()
    geom_col = gdf.geometry.name
    for col in gdf.columns:
        if col == geom_col:
            continue
        if gdf[col].dtype == "object":
            gdf[col] = gdf[col].astype("string")
        elif pd.api.types.is_datetime64_any_dtype(gdf[col]):
            # Drop timezone info — Arrow 15 stores tz-naive timestamps as
            # timestamp[ms] without ambiguity, avoiding schema merge errors
            # when files originate from multiple time zones.
            gdf[col] = gdf[col].dt.tz_localize(None)
    return gdf

#Step 3 — Parallel execution & memory management

Process files concurrently using process-based parallelism. Python’s GIL limits thread-based concurrency for CPU-bound geospatial operations. Use concurrent.futures.ProcessPoolExecutor with a dynamic worker count based on available cores and estimated per-file memory footprint. Each worker operates in a clean namespace, preventing GDAL driver state from leaking between files.

# Python 3.10+  concurrent.futures (stdlib)
import logging
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path

log = logging.getLogger(__name__)


def convert_single_file(filepath: str, output_dir: str) -> dict:
    """Per-worker conversion entry point.

    Each call runs in an isolated process — GDAL state is private.
    Returns a result dict with 'output_path' on success or 'error' on failure.
    """
    import geopandas as gpd
    import pyogrio  # noqa: F401 — ensures GDAL drivers are registered in this worker

    src = Path(filepath)
    dst = Path(output_dir) / src.with_suffix(".parquet").name
    try:
        gdf = gpd.read_file(src, engine="pyogrio")
        gdf = normalize_schema(gdf)
        gdf = validate_and_repair(gdf)
        serialize_to_geoparquet(gdf, dst)
        return {"filepath": filepath, "output_path": str(dst), "error": None}
    except Exception as exc:
        return {"filepath": filepath, "output_path": None, "error": str(exc)}


def run_batch_conversion(
    file_list: list[str],
    output_dir: str,
    max_workers: int | None = None,
) -> tuple[list[dict], list[dict]]:
    """Convert a list of vector files in parallel.

    Returns (successes, failures) as separate lists for downstream routing.
    """
    if max_workers is None:
        max_workers = max(1, mp.cpu_count() - 2)

    successes: list[dict] = []
    failures: list[dict] = []

    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(convert_single_file, f, output_dir): f
            for f in file_list
        }
        for future in as_completed(futures):
            filepath = futures[future]
            try:
                result = future.result()
            except Exception as exc:
                result = {"filepath": filepath, "output_path": None, "error": str(exc)}

            if result.get("error"):
                log.error("conversion failed: %s — %s", filepath, result["error"])
                failures.append(result)
            else:
                log.info("converted: %s → %s", filepath, result["output_path"])
                successes.append(result)

    return successes, failures

#Step 4 — Output serialization & compression

Serialize to GeoParquet using the Arrow engine. Enable ZSTD compression and enforce consistent column ordering. Critical metadata — coordinate reference systems, layer names, and source provenance — must be embedded in the Parquet file schema. The GeoParquet 1.0 specification mandates that CRS and geometry-column metadata be stored in the geo Parquet metadata key; geopandas>=1.0 injects this automatically.

For detailed guidance on embedding audit trails and custom attributes during serialization, see Preserving Metadata During GeoParquet Conversion.

# geopandas>=1.0  pyarrow>=15.0
from pathlib import Path
import geopandas as gpd
import pyarrow as pa
import pyarrow.parquet as pq


def serialize_to_geoparquet(
    gdf: gpd.GeoDataFrame,
    output_path: Path,
    compression: str = "zstd",
    compression_level: int = 3,
    source_filepath: str | None = None,
) -> None:
    """Write a GeoDataFrame as GeoParquet with ZSTD level-3 compression.

    Embeds source provenance in Parquet key-value metadata for lineage
    tracking. geopandas>=1.0 injects spec-compliant GeoParquet `geo`
    metadata automatically; we merge our lineage keys alongside it.

    Writes to a `.tmp` sibling first and renames atomically so a crash
    mid-write never leaves a partial file the row-count check would miss.
    """
    output_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = output_path.with_suffix(output_path.suffix + ".tmp")

    # Build the Arrow table via geopandas so the `geo` key (CRS, geometry
    # encoding) is populated to spec, then extend the schema metadata.
    table = gpd.io.arrow._geopandas_to_arrow(gdf, index=False)
    if source_filepath:
        existing = table.schema.metadata or {}
        merged = {**existing, b"source_filepath": source_filepath.encode()}
        table = table.replace_schema_metadata(merged)

    pq.write_table(
        table,
        tmp_path,
        compression=compression,
        compression_level=compression_level,
    )
    tmp_path.replace(output_path)  # atomic on POSIX filesystems

#Step 5 — Geometry validation & repair

Invalid geometries — self-intersections, unclosed rings, or duplicate vertices — will raise errors during Parquet serialization or produce incorrect spatial join results downstream. Run lightweight topology checks before export.

# shapely>=2.0  geopandas>=1.0
import geopandas as gpd
from shapely.validation import make_valid


def validate_and_repair(
    gdf: gpd.GeoDataFrame,
    drop_null: bool = True,
) -> gpd.GeoDataFrame:
    """Apply Shapely's make_valid; drop irreparable or null geometries.

    Logs a warning per dropped row so the dead-letter audit trail is complete.
    """
    import logging
    log = logging.getLogger(__name__)
    gdf = gdf.copy()

    gdf["geometry"] = gdf["geometry"].apply(
        lambda geom: make_valid(geom) if geom is not None else None
    )

    before = len(gdf)
    valid_mask = gdf.geometry.is_valid
    if drop_null:
        valid_mask &= gdf.geometry.notna()
    dropped = before - valid_mask.sum()
    if dropped:
        log.warning("dropped %d irreparable geometry rows", dropped)

    return gdf.loc[valid_mask].reset_index(drop=True)

#Step 6 — Post-conversion validation & fallback routing

Post-conversion validation is critical for data lake integrity. Verify row counts, geometry type consistency, CRS round-trip fidelity, and checksum matches between source and destination. Implement a fallback routing mechanism that quarantines failed jobs, logs structured error payloads, and triggers retry logic for transient failures such as network timeouts or temporary I/O locks.

The companion guide Fallback Routing for Failed Migration Jobs covers dead-letter queue design, retry back-off strategies, and partial-write rollback in detail.

# geopandas>=1.0  pyogrio>=0.7
import geopandas as gpd
import pyogrio


def validate_output(
    source_path: str,
    output_path: str,
) -> dict:
    """Compare source and output file metadata for basic integrity checks.

    Returns a dict with 'passed' bool and a list of 'failures' (empty if clean).
    """
    src_meta = pyogrio.read_info(source_path)
    dst_gdf = gpd.read_parquet(output_path)

    failures = []

    src_count = src_meta.get("features", 0)
    dst_count = len(dst_gdf)
    if src_count != dst_count:
        failures.append(
            f"row count mismatch: source={src_count}, output={dst_count}"
        )

    src_crs = src_meta.get("crs")
    dst_crs = dst_gdf.crs.to_epsg() if dst_gdf.crs else None
    if src_crs and dst_crs and str(src_crs) != str(dst_crs):
        failures.append(f"CRS mismatch: source={src_crs}, output={dst_crs}")

    return {"passed": len(failures) == 0, "failures": failures}

#Benchmark reference matrix

The table below shows representative conversion outcomes for three common source formats using the pipeline above on a 16-core machine with NVMe scratch storage. Your results will vary with geometry complexity and column width.

ZSTD level 3 is the recommended default for batch pipelines: it achieves 65–75% of maximum possible compression at roughly 3–4× the decompression speed of level 9. Raise to level 6 only for cold-archive outputs where read latency is not a constraint.

For row-group sizing guidance that affects both compression ratio and query-time predicate push-down efficiency, see Row Group Sizing Strategies for Parquet.

#Failure modes & gotchas

GDAL driver state leakage across threads. GDAL registers format drivers in process-global state. Running multiple gdf.read_file() calls in threads sharing a single Python process causes intermittent Unable to open datasource errors at high concurrency. Always use ProcessPoolExecutor; never ThreadPoolExecutor for GDAL workloads.

Silent CRS erasure on Shapefile read. Shapefiles with a malformed or missing .prj sidecar will load with crs=None in geopandas. Downstream to_parquet() will succeed but write a file with no CRS metadata — silently breaking any spatial join. Always check gdf.crs is None immediately after read, and route affected files to the dead-letter queue with an explicit error label.

object-dtype string columns breaking Arrow schema merges. When merging Parquet files written from different source files, Arrow requires identical column types. A single file where a numeric column was read as object (due to a stray string value in the Shapefile DBF) will abort the merge. The normalize_schema() function above prevents this by casting all object columns before write.

Memory exhaustion on large polygon layers. make_valid() on complex multi-polygon geometries (e.g. high-resolution coastline layers with millions of vertices) can transiently allocate 4–8× the apparent geometry memory. Profile with tracemalloc on a representative sample before setting max_workers — overcommitting RAM causes the OS to swap, which destroys throughput.

Partial writes on worker crash. A worker process killed mid-write (OOM, disk full) leaves a partial Parquet file that is not caught by row-count checks. The serialize_to_geoparquet() function above writes to a .tmp sibling and then calls Path.replace() to swap it into place atomically only on successful completion, so an interrupted run leaves a stray .tmp file rather than a corrupt .parquet. Sweep orphaned .tmp files at the start of each run.

CRS mismatch before spatial indexing. If you later build a spatial index — for example to drive quadtree-based spatial partitioning — every file in the batch must share the same CRS. Mixing EPSG:4326 and EPSG:3857 files in one index produces geometrically incorrect results without raising any error. Enforce a single target CRS in the probe step and reproject before serialization.

#Production deployment notes

• I/O tuning: mount cloud storage via s3fs or gcsfs with readahead and block_size set to 64–128 MB. Smaller block sizes produce excessive S3 API calls that inflate latency and cost.
• Orchestration: wrap the pipeline in a DAG runner (Prefect, Airflow, or Dagster) to manage dependencies, retries, and alerting. For high-frequency ingestion, implement event-driven triggers via SQS or Pub/Sub rather than polling.
• CI/CD integration: test the pipeline with synthetic GeoJSON and Shapefile fixtures in CI. Validate outputs against the OGC GeoParquet conformance suite before promoting to production. Automate the validate_output() function as a post-step gate.
• Cost tracking: log compressed output size per file in the success payload. Aggregate across a run to compute cost-per-GB written to object storage, enabling capacity planning and anomaly detection on unusually large outputs.

For a ready-to-deploy script with CI/CD integration guidance, see Automating Shapefile to GeoParquet Conversion.

#FAQ

Why use ProcessPoolExecutor instead of ThreadPoolExecutor for batch GeoParquet conversion?

GDAL geometry operations and Arrow serialization are CPU-bound and hold the Python GIL intermittently. Process-based parallelism gives each worker a separate Python interpreter and GIL, yielding near-linear throughput scaling on multi-core machines. ThreadPoolExecutor only helps for pure I/O waits — not for the CPU-intensive geometry encoding that dominates batch conversion time.

Should I use pyogrio or fiona for reading Shapefiles in a batch pipeline?

Use pyogrio. It exposes a vectorized read path that streams data directly into Arrow buffers, bypassing Python-level feature iteration. For a 1 M-feature Shapefile, pyogrio is typically 4–8× faster than fiona for reading and consumes 30–50% less peak RAM because it avoids materializing intermediate Python dicts per feature.

How do I handle CRS mismatches when converting a mixed-CRS dataset batch?

Probe each file’s CRS with pyogrio.read_info() before conversion. Files that differ from the pipeline’s canonical CRS (e.g. EPSG:4326) should be reprojected with gdf.to_crs() before serialization. Log the source CRS as a Parquet custom metadata key so downstream queries can audit the transformation lineage.

What ZSTD compression level should I use for GeoParquet in a batch pipeline?

Level 3 is the right default: it achieves 65–75% of maximum compression at roughly 3–4× the decompression speed of level 9. Raise to level 6 only for cold-archive outputs where read latency is not critical. The ZSTD compression levels guide provides workload-specific benchmarks for vector and raster data.

#Related

• Automating Shapefile to GeoParquet Conversion — ready-to-deploy script with CI/CD hooks
• Fallback Routing for Failed Migration Jobs — dead-letter queue design and retry back-off
• Preserving Metadata During GeoParquet Conversion — CRS, audit trails, and custom Parquet metadata
• Schema Mapping for Legacy to Modern Formats — type coercion, null handling, nested field flattening
• ZSTD Compression Levels for Geospatial Data — per-level benchmarks for vector and raster workloads

← Back to Data Conversion & Migration Pipelines