Batch Transforming 10k+ Shapefiles Without Memory Leaks Jump to heading#

# Requires: Python >= 3.10
import os

os.environ["GDAL_CACHEMAX"] = "128"                    # MB; hard cap on the block cache
os.environ["OGR_ENABLE_PARTIAL_REPROJECTION"] = "NO"   # fail loudly instead of silently dropping
os.environ["CPL_DEBUG"] = "OFF"                         # suppress per-feature debug allocations
os.environ["VSI_CACHE"] = "FALSE"                       # no virtual file-system read cache
os.environ["GDAL_DISABLE_READDIR_ON_OPEN"] = "EMPTY_DIR"  # skip sidecar directory scans

# schema_mapping.yaml
target_schema:
  - name: parcel_id
    type: str
    source_fields: ["PARCEL_ID", "PID", "APN"]
  - name: land_use_code
    type: str
    source_fields: ["LU_CODE", "LANDUSE", "ZONING"]
  - name: area_sqm
    type: float
    source_fields: ["AREA", "SQ_METERS", "ACRES"]
    transform: "lambda x: x * 4046.86 if x else None"   # acres -> square metres
validation:
  max_field_mismatch_pct: 15
  required_geometry: "Polygon"

# Requires: Python >= 3.10, PyYAML >= 6.0
import yaml
from typing import Any


def load_schema_config(config_path: str) -> dict[str, Any]:
    with open(config_path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


def validate_schema_match(src_props: dict[str, str], config: dict[str, Any]) -> bool:
    """Return True if matched target fields satisfy the mismatch threshold."""
    target_fields = config["target_schema"]
    source_fields = set(src_props.keys())

    matched = sum(
        1 for f in target_fields
        if any(s in source_fields for s in f["source_fields"])
    )
    mismatch_pct = (1 - (matched / len(target_fields))) * 100
    return mismatch_pct <= config["validation"]["max_field_mismatch_pct"]

# Requires: Python >= 3.10, fiona >= 1.9, pyproj >= 3.6
import gc
import resource
import logging
from pathlib import Path
from typing import Any

import fiona
from fiona.transform import transform_geom
from pyproj import CRS

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("shapefile_etl")

TARGET_CRS = "EPSG:4326"
TARGET_CRS_OBJ = CRS.from_epsg(4326)


def get_rss_mb() -> float:
    """Return peak RSS in MB. ru_maxrss is KB on Linux, bytes on macOS."""
    return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024


def transform_record(feat: dict[str, Any], config: dict[str, Any]) -> dict[str, Any]:
    """Apply field mapping and type coercion to one feature's properties."""
    mapped: dict[str, Any] = {}
    for field in config["target_schema"]:
        value = None
        for src in field["source_fields"]:
            if src in feat["properties"]:
                value = feat["properties"][src]
                break

        if "transform" in field and value is not None:
            value = eval(field["transform"])(value)  # noqa: S307 — trusted manifest only

        if field["type"] == "float" and value is not None:
            try:
                value = float(value)
            except (ValueError, TypeError):
                value = None
        elif field["type"] == "str":
            value = str(value) if value is not None else None

        mapped[field["name"]] = value
    return mapped


def process_batch(
    file_batch: list[Path],
    config: dict[str, Any],
    output_dir: Path,
    quarantine_dir: Path,
) -> None:
    """Process one chunk of shapefiles with explicit teardown per file."""
    for src_file in file_batch:
        try:
            with fiona.open(str(src_file), "r") as src:
                if not validate_schema_match(src.schema["properties"], config):
                    raise ValueError("Schema mismatch exceeds threshold")
                if src.crs is None:
                    raise RuntimeError("Missing .prj file — CRS undefined")

                src_crs_obj = CRS.from_user_input(src.crs)
                needs_transform = not src_crs_obj.equals(TARGET_CRS_OBJ)

                out_schema = {
                    "geometry": config["validation"]["required_geometry"],
                    "properties": {f["name"]: f["type"] for f in config["target_schema"]},
                }
                out_path = output_dir / f"{src_file.stem}_standardized.shp"

                with fiona.open(
                    str(out_path), "w",
                    driver="ESRI Shapefile",
                    schema=out_schema,
                    crs=TARGET_CRS,
                ) as dst:
                    for feat in src:
                        geom = feat["geometry"]
                        if geom is None or not geom.get("coordinates"):
                            continue  # skip empty geometries rather than writing nulls
                        if needs_transform:
                            geom = transform_geom(src.crs, TARGET_CRS, geom)
                        mapped_props = transform_record(feat, config)
                        dst.write({"type": "Feature", "geometry": geom, "properties": mapped_props})

            logger.info("Completed: %s", src_file.name)

        except Exception as e:  # quarantine, never crash the batch
            (quarantine_dir / src_file.name).write_bytes(src_file.read_bytes())
            src_file.unlink()
            logger.error("Quarantined %s: %s", src_file.name, e)
        finally:
            gc.collect()  # reclaim cyclic refs left by the fiona session
            rss = get_rss_mb()
            if rss > 1500:  # soft GC trigger from Step 1
                logger.warning("High RSS (%.0f MB) — forcing second GC pass.", rss)
                gc.collect()

# Requires: Python >= 3.10
def main() -> None:
    config = load_schema_config("schema_mapping.yaml")
    input_dir = Path("input_shapefiles")
    output_dir = Path("output_standardized")
    quarantine_dir = Path("quarantine")
    output_dir.mkdir(exist_ok=True)
    quarantine_dir.mkdir(exist_ok=True)

    files = sorted(input_dir.glob("*.shp"))
    batch_size = 75

    for i in range(0, len(files), batch_size):
        batch = files[i : i + batch_size]
        logger.info("Batch %d (%d files) — RSS %.0f MB", i // batch_size + 1, len(batch), get_rss_mb())
        process_batch(batch, config, output_dir, quarantine_dir)

    logger.info("Batch transformation complete: %d files processed.", len(files))


if __name__ == "__main__":
    main()

# Confirm RSS never crossed the 2.0 GB hard ceiling across the full run
grep "Batch " etl.log | awk -F 'RSS ' '{print $2}' | sort -n | tail -1
# Expected: a value well under 2048 MB, and roughly equal to the first batch's RSS

# Requires: Python >= 3.10, pytest >= 7
# A fixture-style assertion you can run after a 10k batch in CI.
from pathlib import Path


def test_no_silent_drops(input_count: int) -> None:
    standardized = len(list(Path("output_standardized").glob("*_standardized.shp")))
    quarantined = len(list(Path("quarantine").glob("*.shp")))
    # Every input file is accounted for — nothing vanished silently.
    assert standardized + quarantined == input_count

INFO  Batch 1 (75 files) — RSS 412 MB
INFO  Batch 134 (75 files) — RSS 419 MB        # baseline barely moved over 10k files
WARNING High RSS (1521 MB) — forcing second GC pass.
ERROR Quarantined county_2009_legacy.shp: Missing .prj file — CRS undefined