Geospatial Schema Standardization & Mapping Automation

Flattening Deeply Nested GeoJSON Feature Collections Safely

Geospatial ETL pipelines frequently break when ingesting vendor-generated GeoJSON containing arbitrary nesting levels, mixed-type arrays, and unbounded metadata objects. Flattening these payloads safely requires deterministic recursion boundaries, strict coordinate precision management, and explicit type coercion rules. Without these controls, downstream spatial databases encounter schema drift, geometry validation failures, and silent data truncation. This page is the runnable implementation companion to the Nested JSON/GeoJSON Flattening stage, itself a high-precision transformation step within Automated Attribute Transformation & ETL Workflows.

The hard part is not the recursion itself — it is making the recursion bounded and reproducible so that the same vendor file always produces the same column set. Government and enterprise GIS systems expect rigid attribute schemas; a feature collection that nests an owner object three levels deep on Monday and four levels deep on Tuesday must still emit a predictable, stable record. The engine below enforces that contract, and because it holds no per-run state it drops cleanly into batch schema processing pipelines that stream thousands of files without accumulating memory.

Bounded GeoJSON Flattening Pipeline A nested FeatureCollection enters four sequential stages — configure boundaries, execute bounded flattening, validate the schema, and log lineage. The attribute properties travel along the main stage path and are projected into flat columns, while the geometry coordinate array follows a separate lower lane that bypasses attribute flattening, is only precision-clamped, and rejoins the validated flat record on the right. Nested FeatureCollection properties{ } geometry[ ] 1 - Configure max_depth, separator precision, type rules 2 - Execute bounded recursion dot-notation keys 3 - Validate OGC geom type column-name regex 4 - Log OK / QUARANTINED / FAILED, lineage attributes flattened to columns geometry bypass precision-clamp only coordinates never become attribute columns Validated Flat Record owner__id: 1043 geometry: clamped

Core Failure Modes in Nested GeoJSON

When a feature collection contains deeply nested objects or coordinate-like strings, naive flattening produces unpredictable column names, type collisions, and geometry corruption. The most frequent pipeline failures are:

  • ValueError: invalid literal for float() during coordinate or attribute parsing
  • SchemaMismatch during batch inserts into PostGIS or GeoPackage when a new nesting level invents an unexpected column
  • Silent precision loss when floating-point coordinates exceed database decimal limits
  • Unhandled None geometries triggering topology validation crashes

Resolving these requires separating spatial geometry from attribute flattening, enforcing an explicit recursion depth, and applying deterministic key generation. The same boundary discipline used in robust field type casting scripts applies here: a failed coercion should raise or be logged, never be silently swallowed.

Prerequisites

Before running the flattener, confirm your environment meets these requirements:

Step 1 — Configure Flattening Boundaries and Type Rules

Isolate the flattening configuration from execution logic. A minimal, reproducible config establishes maximum recursion depth, coordinate precision, and explicit type mapping. This prevents unbounded traversal of vendor metadata and guarantees consistent column naming across runs.

python
# flatten_config.py
# Requires: Python >= 3.10
from typing import Any

FLATTEN_CONFIG: dict[str, Any] = {
    "max_depth": 4,              # cap recursion — sub-objects below this are serialised
    "separator": "__",          # deterministic key joiner: owner__id, owner__contact__email
    "coordinate_precision": 6,  # decimal places ~= 0.1 m at the equator (WGS84)
    "type_coercion": {
        "bool": ["true", "false", "1", "0"],
        "int": ["integer", "count", "total"],
        "float": ["decimal", "ratio", "coordinate", "lat", "lon"],
        "string": "default",
    },
    "preserve_keys": ["id", "geometry", "properties"],
    "drop_nulls": True,         # avoid sparse columns in columnar stores
    "default_crs": "EPSG:4326",
}

Apply these thresholds during pipeline initialisation:

  • Recursion depth: Cap at 4 levels. This covers the overwhelming majority of municipal and federal GeoJSON payloads without risking stack overflow or runaway column explosion. Anything deeper is serialised to a single JSON-text column.
  • Coordinate precision: Set to 6 decimal places. Aligns with ~0.1 m WGS84 accuracy and prevents floating-point bloat in spatial indexes. If your downstream CRS is metric, validate the choice against CRS Normalization & Sync before lowering it.
  • Type coercion: Map key-name hints to explicit Python types before database insertion, eliminating downstream casting errors.
  • Null handling: Drop None values by default to prevent sparse column generation in columnar stores.
  • CRS enforcement: Validate or default to EPSG:4326 per RFC 7946 compliance.

Step 2 — Execute Bounded Flattening with Precision Management

Implement a deterministic engine that separates geometry processing from attribute traversal. The class below handles recursion, type coercion, and precision clamping in a single pass. Coordinate arrays are clamped but never flattened into columns — the geometry passes through structurally intact.

python
# geojson_flattener.py
# Requires: Python >= 3.10
import json
import logging
from typing import Any

logger = logging.getLogger(__name__)


class GeoJSONFlattener:
    def __init__(self, config: dict[str, Any]):
        self.max_depth = config.get("max_depth", 4)
        self.separator = config.get("separator", "__")
        self.precision = config.get("coordinate_precision", 6)
        self.type_coercion = config.get("type_coercion", {})
        self.drop_nulls = config.get("drop_nulls", True)
        self.default_crs = config.get("default_crs", "EPSG:4326")

    def _coerce_value(self, key: str, value: Any) -> Any:
        if value is None:
            return None
        if isinstance(value, bool):
            return value
        if isinstance(value, (int, float)):
            return value

        val_str = str(value).strip().lower()
        if val_str in self.type_coercion.get("bool", []):
            return val_str in ("true", "1")
        if any(k in key.lower() for k in self.type_coercion.get("int", [])):
            try:
                return int(float(value))
            except (ValueError, TypeError):
                pass
        if any(k in key.lower() for k in self.type_coercion.get("float", [])):
            try:
                return round(float(value), self.precision)
            except (ValueError, TypeError):
                pass
        return str(value)

    def _flatten_dict(self, obj: dict, parent_key: str = "", depth: int = 0) -> dict:
        items: dict[str, Any] = {}
        if depth >= self.max_depth:
            # Serialise the entire sub-object once the depth limit is reached
            return {parent_key: json.dumps(obj, ensure_ascii=False)} if parent_key else {}

        for k, v in obj.items():
            new_key = f"{parent_key}{self.separator}{k}" if parent_key else k
            if isinstance(v, dict):
                items.update(self._flatten_dict(v, new_key, depth + 1))
            elif isinstance(v, list):
                if all(isinstance(x, (str, int, float, bool, type(None))) for x in v):
                    items[new_key] = [self._coerce_value(new_key, x) for x in v]
                else:
                    items[new_key] = json.dumps(v, ensure_ascii=False)
            else:
                items[new_key] = self._coerce_value(new_key, v)
        return items

    def _clamp_precision(self, coords: Any) -> Any:
        if isinstance(coords, list):
            return [self._clamp_precision(c) for c in coords]
        if isinstance(coords, (int, float)):
            return round(float(coords), self.precision)
        return coords

    def process_feature(self, feature: dict) -> dict:
        if not isinstance(feature, dict) or feature.get("type") != "Feature":
            raise ValueError("Invalid GeoJSON Feature structure")

        geom = feature.get("geometry")
        if geom is None or not isinstance(geom, dict):
            # Preserve a null-geometry marker rather than fabricating coordinates
            geom = None
        else:
            geom = dict(geom)
            geom["coordinates"] = self._clamp_precision(geom.get("coordinates"))

        props = feature.get("properties") or {}
        flat_props = self._flatten_dict(props)

        if self.drop_nulls:
            flat_props = {k: v for k, v in flat_props.items() if v is not None}

        return {
            "feature_id": feature.get("id"),
            "geometry_type": geom.get("type") if geom else None,
            "geometry": geom,
            **flat_props,
        }

The two key invariants: depth-limited recursion guarantees a bounded, reproducible column set, and _clamp_precision keeps geometry numerically stable without ever promoting coordinates into attribute columns.

The depth boundary is what stops a vendor’s variable nesting from rewriting your schema. Keys above the max_depth line each become their own flat column; the moment traversal reaches the limit, the remaining sub-object is serialised whole into a single JSON-text column — so one extra nesting level can never invent new columns on some files but not others.

How the max_depth Boundary Stabilises the Column Set A nested properties object is shown as a vertical ladder of depth levels 0 through 4. Each key above the max_depth equals 4 boundary line (name at depth 0, owner.id and owner.contact.email across depths 1 to 3) is flattened into its own deterministic column using the double-underscore separator. At depth 4 the boundary is crossed: the remaining sub-object is not expanded into columns but serialised once into a single JSON-text column, so deeper or variable nesting cannot add or remove columns between files. Nested properties{ } name: "Lot 7" d0 owner: { } d1 id: 1043 d2 contact: { } d3 history: { … } audit: { … } d4 max_depth = 4 boundary Flat, stable columns name owner__id owner__contact__email owner__history one JSON-text column — sub-object serialised whole above the boundary: one column per key at / below the boundary: collapsed, so the schema never drifts

Step 3 — Validate the Flattened Schema

Flattened output must align with enterprise spatial standards before it touches a database. Enforce these checks immediately after process_feature and quarantine anything that fails rather than inserting it.

python
# validate_record.py
# Requires: Python >= 3.10
import re

OGC_SIMPLE_FEATURES = {
    "Point", "LineString", "Polygon",
    "MultiPoint", "MultiLineString", "MultiPolygon",
}
COLUMN_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")


def validate_record(record: dict) -> list[str]:
    """Return a list of validation failures; empty list means the record is clean."""
    failures: list[str] = []

    gtype = record.get("geometry_type")
    if gtype is None:
        failures.append("null_geometry")  # quarantine, never insert
    elif gtype not in OGC_SIMPLE_FEATURES:
        failures.append(f"non_ogc_geometry:{gtype}")

    for key in record:
        if key in ("geometry",):
            continue
        if not COLUMN_RE.match(key):
            failures.append(f"illegal_column_name:{key}")

    return failures

Apply these rules:

  • Geometry validation: geometry_type must map to a valid OGC Simple Features type (Point, LineString, Polygon, MultiPolygon, and their Multi* siblings). Null geometries are quarantined, not inserted.
  • Column naming: Flattened keys must match [a-zA-Z_][a-zA-Z0-9_]* to avoid SQL reserved-keyword collisions and injection-prone identifiers.
  • Precision consistency: Every coordinate array must already be clamped to the configured decimal threshold from Step 2 before export.
  • Auditability: Retain the original id and geometry_type columns for lineage tracing through the rest of the workflow.

For a final geometry gate, validate output against ST_IsValid() in PostGIS or shapely’s is_valid before deployment — this guarantees interoperability across municipal, state, and federal GIS platforms.

Step 4 — Log Lineage and Route Batch Failures

Production pipelines must survive malformed inputs and CRS drift without halting. Wrap batch processing so each feature either produces a clean record or a standardised error record — no silent failures, no aborted batches. This mirrors the failure-routing discipline in Error Handling & Retry Logic: transient infrastructure faults retry, but a malformed feature is logged and skipped.

python
# process_batch.py
# Requires: Python >= 3.10
import logging
from geojson_flattener import GeoJSONFlattener
from validate_record import validate_record

logger = logging.getLogger(__name__)


def process_batch(flattener: GeoJSONFlattener, features: list[dict]) -> list[dict]:
    results: list[dict] = []
    for idx, feat in enumerate(features):
        try:
            # RFC 7946 §4 deprecates the top-level "crs" member; warn on anything
            # other than CRS84 so non-compliant payloads surface in CI logs.
            crs = feat.get("crs", {})
            if crs:
                crs_name = crs.get("properties", {}).get("name", "")
                if crs_name not in ("urn:ogc:def:crs:OGC:1.3:CRS84", ""):
                    logger.warning(
                        "Feature %d: non-standard CRS '%s' — expecting WGS84 (CRS84)",
                        idx, crs_name,
                    )

            record = flattener.process_feature(feat)
            failures = validate_record(record)
            if failures:
                results.append({
                    "feature_id": record.get("feature_id", f"unknown_{idx}"),
                    "status": "QUARANTINED",
                    "violations": failures,
                })
            else:
                record["status"] = "OK"
                results.append(record)
        except Exception as exc:
            logger.error("Feature %d failed flattening: %s", idx, exc)
            results.append({
                "feature_id": feat.get("id", f"unknown_{idx}"),
                "error": str(exc),
                "status": "FAILED",
            })
    return results

The audit trail written by this stage should record, per feature: the original feature_id, the final status (OK / QUARANTINED / FAILED), and any violations or error text. That record set is what downstream lineage and rejection-log reporting consumes.

Verification

Confirm the engine behaves deterministically before wiring it into a pipeline. The test below asserts that a four-level-deep object is serialised (not exploded into columns), coordinates are clamped, and a null geometry is quarantined.

python
# test_flattener.py
# Requires: Python >= 3.10, pytest >= 7
from flatten_config import FLATTEN_CONFIG
from geojson_flattener import GeoJSONFlattener
from process_batch import process_batch


def test_bounded_flatten_and_quarantine():
    flattener = GeoJSONFlattener(FLATTEN_CONFIG)
    features = [
        {
            "type": "Feature",
            "id": "lot-7",
            "geometry": {"type": "Point", "coordinates": [-73.987654321, 40.748817123]},
            "properties": {"name": "Lot 7", "owner": {"id": 1043, "contact": {"email": "[email protected]"}}},
        },
        {"type": "Feature", "id": "lot-8", "geometry": None, "properties": {"name": "Lot 8"}},
    ]
    out = process_batch(flattener, features)

    ok = out[0]
    assert ok["status"] == "OK"
    assert ok["geometry"]["coordinates"] == [-73.987654, 40.748817]  # clamped to 6 dp
    assert ok["owner__id"] == 1043                                    # nested key joined

    quarantined = out[1]
    assert quarantined["status"] == "QUARANTINED"
    assert "null_geometry" in quarantined["violations"]

Run it with:

bash
pytest test_flattener.py -v

Expected output:

text
test_flattener.py::test_bounded_flatten_and_quarantine PASSED

In production runs, watch the structured log sink for the CRS warning to confirm non-compliant payloads are being caught rather than silently ingested:

text
WARNING Feature 12: non-standard CRS 'urn:ogc:def:crs:EPSG::2263' — expecting WGS84 (CRS84)

Troubleshooting

Symptom Likely cause Fix
New columns appear on some files but not others A vendor occasionally nests one level deeper, exploding into extra keys Lower max_depth so the variable sub-object is serialised to one JSON-text column, stabilising the schema
ValueError: invalid literal for float() on a value that looks numeric A key-name hint (lat, coordinate) matched a free-text field Tighten the type_coercion hint lists, or coerce only on exact key names rather than substring matches
Coordinates lose meaningful precision after flattening coordinate_precision is too low for your target CRS Raise the precision; for projected/metric CRSs confirm the right value via Unit Conversion Tolerance Thresholds before exporting
Batch insert fails with reserved-keyword column error A flattened key (e.g. order, user) collides with SQL reserved words The validator flags illegal names; add a prefix/quoting rule before insert, or rename via field renaming and type coercion rules
Records silently disappear from output Geometry was None and the feature was quarantined, not inserted Inspect the QUARANTINED records in the audit trail — null geometries are intentionally held back, not dropped