Spatial Graph Construction & OSM Ingestion

When an OpenStreetMap extract is loaded naively — one node per coordinate, one transaction per way — a metropolitan region will fragment the heap, bloat the transaction log, and leave the routing graph riddled with phantom edges and disconnected subgraphs. Backend and data engineers building logistics, mobility, or geospatial-analytics systems need an ingestion architecture that produces a deterministic, directed, weighted edge network on the first pass and keeps it correct as the map changes. This guide covers the full construction path: stream deserialization, topology resolution, schema design, the async Neo4j integration layer, spatial indexing and query planning, traversal-readiness, and the failure modes that decide whether the pipeline survives contact with production.

The pipeline decomposes into three resource-bounded stages — stream deserialization, topology resolution, and graph materialization — each running under explicit concurrency and memory limits so that a single large extract never exhausts the driver connection pool or saturates the write-ahead log.

Three-stage OSM-to-graph ingestion pipeline A left-to-right pipeline of three resource-bounded stages. Stage one, stream deserialization, runs outside any transaction and is CPU and I/O bound: PBF blocks flow into a tag filter with WGS84 projection. Stage two, topology resolution, is pure in-memory work over a bounded geographic partition: coordinate snapping at 0.3 to 0.5 metre tolerance feeds directional edge construction from oneway tags and turn restrictions. Stage three, graph materialization, is the only stage that touches the database, in fixed-size batches: UNWIND batches of up to 50,000 nodes and 200,000 edges write into the routing graph. Arrows connect the stages in order. 1 Stream deserialization outside any transaction · CPU + I/O bound PBF blocks streamed, not buffered whole Tag filter + WGS84 pre-filter ways, project points 2 Topology resolution in-memory · bounded partition Coordinate snapping 0.3–0.5 m tolerance Directional edges oneway + turn restrictions 3 Graph materialization only DB writer · fixed batches UNWIND batches ≤50k nodes / ≤200k edges Routing graph

Concept & Architecture

A road network is a graph before it is anything else: intersections are nodes, the segments between them are edges, and the cost of traversal lives on the edge. Modelling this in a relational store forces every routing query into recursive self-joins against an adjacency table, where each hop is a fresh index probe and the planner has no notion of locality. A native graph store traverses by pointer-following adjacency, so the cost of expanding a path is proportional to the number of edges visited rather than the size of the table — which is exactly the access pattern shortest-path and nearest-neighbour queries need. The trade-off is that the graph must be constructed correctly up front; there is no query-time join to paper over a missing edge.

The storage model rests on two spatial primitives. Nodes carry a location property typed as a WGS84 point (point({latitude, longitude})), which Neo4j stores in a dedicated point index and serves to bounding-box and distance predicates without scanning. Edges carry the traversal cost — distance, traversal time, and directional flags — as scalar properties so the cost function is a property read rather than a runtime computation. The mechanics of turning raw geometry into these primitives are covered in depth in node and edge spatial mapping, and the index structures that make point predicates cheap are the subject of spatial indexing strategies.

Architecturally, the three stages are deliberately decoupled. Stream deserialization is CPU- and I/O-bound and runs outside any transaction. Topology resolution is pure in-memory computation over a bounded geographic partition. Only graph materialization touches the database, and it does so in fixed-size batches. Keeping the database off the critical path until the last stage means a parsing or snapping bug never leaves a half-written graph, and it lets each stage scale on its own resource axis.

Schema Design

The node/edge property model is intentionally narrow: a wide core schema is what causes write amplification at metropolitan scale. The routing graph carries only what the cost function and the indexes need; everything else (demographics, curb access, charging metadata) attaches as separate labels or lives on adjacent nodes so it never inflates a traversal frontier.

// Uniqueness + existence constraints (these implicitly create the backing index)
CREATE CONSTRAINT node_id_unique IF NOT EXISTS
FOR (n:Node) REQUIRE n.id IS UNIQUE;

CREATE CONSTRAINT node_tenant_present IF NOT EXISTS
FOR (n:Node) REQUIRE n.tenant_id IS NOT NULL;

// Point index drives bbox + distance predicates during construction and routing
CREATE POINT INDEX node_location IF NOT EXISTS
FOR (n:Node) ON (n.location);

// Range index supports tenant-scoped scans and partition replays
CREATE INDEX node_tenant IF NOT EXISTS
FOR (n:Node) ON (n.tenant_id);

Each :Node holds id (the OSM node id, stable across imports), location (WGS84 point), osm_type, and tenant_id. Each CONNECTS_TO relationship is directional and holds distance_m, traversal_s (the normalised cost), max_speed_kph, and surface. Relationship direction is load-bearing: an OSM way tagged oneway=yes materialises as a single (a)-[:CONNECTS_TO]->(b), a default bidirectional way materialises as two opposing relationships, and oneway=-1 reverses the pair. Storing direction explicitly — rather than as a boolean the query must interpret — lets the traversal engine prune at expansion time.

Tenant isolation is enforced by carrying tenant_id on every node and partitioning indexes accordingly, so one customer’s road network can never bleed into another’s routing result. The broader access-control model — label scoping, per-tenant index partitions, and query-time guards — is the remit of spatial security boundaries, and should be wired in before the first production ingest rather than retrofitted.

Core Python Integration

Ingestion uses the official neo4j async driver (Python 3.10+). The connection pool is sized once, sessions are scoped to a single partition’s batch, and every write goes through UNWIND so a batch is one round-trip rather than thousands. The example below is self-contained and runnable: it computes Haversine distances for edge weights, materialises nodes and directional edges in capped batches, and drives the whole thing from an asyncio entry point.

import asyncio
import math
from typing import Dict, List
from neo4j import AsyncGraphDatabase
from neo4j.exceptions import Neo4jError, ServiceUnavailable

EARTH_RADIUS_M = 6_371_000.0

def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Great-circle distance in metres, used to weight edges during materialization."""
    phi1, phi2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlambda = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
    return 2 * EARTH_RADIUS_M * math.asin(math.sqrt(a))


class GraphIngestor:
    """Async OSM-to-graph materializer with bounded pool and capped batches."""

    NODE_CAP = 50_000
    EDGE_CAP = 200_000

    def __init__(self, uri: str, auth: tuple, database: str = "spatial_routing",
                 max_pool: int = 16):
        self._driver = AsyncGraphDatabase.driver(
            uri,
            auth=auth,
            max_connection_pool_size=max_pool,
            connection_acquisition_timeout=10.0,
            max_transaction_retry_time=30.0,
        )
        self._database = database

    async def close(self) -> None:
        await self._driver.close()

    async def materialize(self, batch: List[Dict]) -> int:
        """Commit one partition batch: MERGE nodes, then directional edges."""
        if len(batch) > self.NODE_CAP:
            raise ValueError(f"batch exceeds node cap ({len(batch)} > {self.NODE_CAP})")
        query = """
        UNWIND $batch AS item
        MERGE (n:Node {id: item.node_id})
          SET n.location  = point({latitude: item.lat, longitude: item.lon}),
              n.osm_type   = item.type,
              n.tenant_id  = item.tenant_id
        WITH n, item
        WHERE item.source_id IS NOT NULL
        MATCH (m:Node {id: item.source_id})
        MERGE (m)-[r:CONNECTS_TO]->(n)
          SET r.distance_m    = item.dist_m,
              r.traversal_s    = item.traversal_s,
              r.max_speed_kph  = item.max_speed_kph,
              r.surface        = item.surface
        RETURN count(r) AS edges
        """
        async with self._driver.session(database=self._database) as session:
            try:
                result = await session.run(query, batch=batch)
                record = await result.single()
                await result.consume()
                return record["edges"] if record else 0
            except (Neo4jError, ServiceUnavailable) as exc:
                raise RuntimeError(f"batch commit failed: {exc}") from exc


def build_edge(src: Dict, dst: Dict, max_speed_kph: float, surface: str,
               tenant_id: str) -> Dict:
    """Derive an edge record with distance and normalised traversal time."""
    dist_m = haversine_distance(src["lat"], src["lon"], dst["lat"], dst["lon"])
    speed_mps = max(max_speed_kph, 5.0) / 3.6
    return {
        "node_id": dst["id"], "lat": dst["lat"], "lon": dst["lon"],
        "type": "junction", "tenant_id": tenant_id,
        "source_id": src["id"], "dist_m": dist_m,
        "traversal_s": dist_m / speed_mps,
        "max_speed_kph": max_speed_kph, "surface": surface,
    }


async def main() -> None:
    ingestor = GraphIngestor("bolt://localhost:7687", ("neo4j", "password"))
    try:
        a = {"id": 1, "lat": 52.5200, "lon": 13.4050}
        b = {"id": 2, "lat": 52.5210, "lon": 13.4065}
        batch = [build_edge(a, b, max_speed_kph=50.0, surface="asphalt",
                            tenant_id="acme")]
        edges = await ingestor.materialize(batch)
        print(f"committed {edges} edge(s)")
    finally:
        await ingestor.close()


if __name__ == "__main__":
    asyncio.run(main())

The pool is bounded (max_connection_pool_size=16) and acquisition is timed out (connection_acquisition_timeout=10.0) so a backed-up database surfaces as a fast, retryable failure rather than a stalled event loop. Sessions are scoped with async with to a single batch, which keeps transaction lifetimes short and lock windows narrow. The production-grade streaming front end — concurrent PBF block parsing, tag pre-filtering, and back-pressured batch hand-off — is detailed in OSM data ingestion pipelines, and the concurrency control that fans these batches across worker pools is covered in async batch processing for graphs.

Indexing & Query Planning

A point index is the difference between a construction pass that finishes in minutes and one that re-scans the node set for every MERGE. During ingestion the MERGE (n:Node {id: ...}) on a unique-constrained id is an exact index seek, and the location point index lets the snapping and validation queries answer bounding-box questions without a label scan. The right index family per access pattern — point index for distance and bbox, range index for tenant scoping, composite where both apply — is analysed in spatial indexing strategies.

The cost model the planner applies to a distance predicate is, at its core, the Haversine great-circle distance between two points on a sphere of radius $R$:

$$d = 2R \cdot \arcsin!\left(\sqrt{\sin^{2}!\frac{\varphi_2-\varphi_1}{2} + \cos\varphi_1\cos\varphi_2,\sin^{2}!\frac{\lambda_2-\lambda_1}{2}}\right)$$

Evaluating this per candidate is cheap; evaluating it per node in the graph is not. The planner avoids the latter through predicate push-down: a point.distance(...) < r filter is rewritten into a bounding-box seek against the point index, which returns a small candidate set, and only those candidates pay for the exact distance computation. When a query mixes Cartesian and WGS84 points, or compares an indexed property against a computed value, that push-down is lost and the planner falls back to a full scan. Validating the plan with PROFILE and shaping it with index hints is the subject of graph query planner optimization, and the query-side patterns that exploit the index live under Cypher spatial queries & pathfinding patterns.

The three-panel diagram below traces a single point.distance(p, q) < r predicate as the planner narrows it: the full node set is never scanned, the point index returns a small bounding-box candidate set, and only those candidates pay for the exact great-circle test.

Bounding-box predicate push-down for a distance filter Three panels left to right showing how a distance predicate is evaluated. The left panel, the full node set, shows many scattered nodes that the planner never scans. The middle panel shows the point index returning only the nodes inside a dashed bounding box around the query point — a small candidate set. The right panel applies the exact distance test: a circle of radius r centred on the query point, where only candidates inside the circle survive as results, shown highlighted, while bbox candidates outside the circle are discarded. Funnel arrows between panels show the set shrinking from the whole graph to a few candidates to the final survivors. Full node set never scanned Point-index bbox seek q bbox candidates Exact distance < r r q exact survivors seek filter

Routing & Traversal Patterns

Construction exists to serve traversal, so the edge schema must match the algorithm family the workload will run. Three families dominate, and the graph is built to suit whichever is dominant.

  • Dijkstra expands the cheapest frontier first and needs only a non-negative traversal_s weight. It is the default for one-to-many queries (isochrones, service-area coverage) where there is no single target to aim at. Build cost: nothing beyond clean, non-negative edge weights.
  • A* prunes Dijkstra’s frontier with an admissible heuristic — typically straight-line Haversine distance to the target divided by the network’s maximum speed, which provably never overestimates remaining cost. It wins on point-to-point queries over large graphs. Build cost: the heuristic needs node location populated and trustworthy, which makes coordinate accuracy during snapping a routing concern, not just a cosmetic one.
  • Contraction hierarchies precompute shortcut edges so query-time expansion skips through unimportant nodes. They deliver the lowest query latency on near-static continental graphs, at the price of a heavy preprocessing pass and shortcut storage. Build cost: highest — and the precompute must be re-run when topology changes materially.

The decision is a function of query shape and update frequency: point-to-point over a stable map favours contraction hierarchies; mixed one-to-many over a frequently edited map favours Dijkstra or A*. Concrete weighting and expansion code — including k-nearest-neighbour routing and distance-filter query patterns — lives in the query guides, but the requirement to keep traversal_s non-negative and location accurate is owned here, at construction time.

Performance & Scale

Memory pressure during construction scales linearly with coordinate density, and the only durable defence is spatial partitioning. Geohash-prefix sharding or a quadtree filter isolates each partition to a fixed geographic extent, so worker pools ingest disjoint regions in parallel without cross-node lock contention, and each worker’s working set stays inside a predictable RAM budget. Snapping then runs over one partition at a time:

def snap_junctions(coords: List[tuple], tolerance_m: float = 0.5) -> List[tuple]:
    """Merge coordinates within tolerance using grid-based hashing (per partition)."""
    grid: Dict[tuple, tuple] = {}
    snapped: List[tuple] = []
    for lat, lon in coords:
        delta = tolerance_m / 111_320.0          # approx metres-per-degree at equator
        key = (round(lat / delta), round(lon / delta))
        if key not in grid:
            grid[key] = (lat, lon)
            snapped.append((lat, lon))
    return snapped

The recurring scale trade-offs:

  • Streaming vs. in-memory: iterating PBF blocks keeps the RAM footprint flat but re-parses tags repeatedly; pre-filtering tags at the parser level recovers most of the CPU cost while preserving the memory win.
  • Snapping tolerance vs. accuracy: aggressive snapping (>1.0 m) collapses distinct parallel carriageways into one edge; conservative snapping (<0.2 m) leaves micro-gaps that break A* and contraction-hierarchy expansion. The 0.3–0.5 m band is the production sweet spot.
  • Batch size vs. write-ahead log: larger batches cut round-trip latency but risk WAL saturation. The 50k-node / 200k-edge caps align with default Neo4j transaction-log rotation thresholds and bound rollback cost.
  • Write amplification: every extra core property multiplies per-edge write volume. Keep the routing graph narrow and attach optional context separately — the model used by POI enrichment workflows, which hangs delivery zones and curb metadata off adjacent nodes without bloating the traversal frontier.

Generator-based batching and memory-mapped buffers keep GC pauses short: by never materialising a whole regional extract as live objects, the collector’s young generation stays small and pauses stay sub-millisecond even mid-ingest.

Failure Modes & Hardening

Topology corruption is the most common silent failure: a snapping tolerance that is too loose welds separate roads together, and one too tight leaves disconnected subgraphs that routing reports as “no path”. Guard it by asserting, post-batch, that every materialised node has a non-zero degree and that the largest connected component covers the expected share of the partition — Haversine validation of each edge length against its claimed distance_m catches the rest.

Index fragmentation accrues under heavy concurrent writes and degrades point-seek latency until background compaction completes. Throttle write concurrency to the index partition count, and warm the point index after a bulk load before serving routing traffic so the first production queries do not pay the cold-cache penalty.

Connection pool exhaustion appears as connection_acquisition_timeout errors when batch fan-out outpaces the pool. Treat the bounded pool as a feature: pair it with a semaphore that caps in-flight batches at the pool size, exponential backoff on retryable errors, and a circuit breaker that sheds load during database maintenance windows.

Recovery playbook: commit at partition boundaries, never across them — a cross-partition transaction turns one bad batch into a distributed rollback. Log each committed partition to an immutable checkpoint ledger so a failed run resumes from the last good partition rather than restarting. During schema migrations or regional outages, replay from those checkpoints; because materialization is idempotent (MERGE, not CREATE), replaying a partition is safe. Keeping the graph current as the underlying map changes — applying road closures and seasonal restrictions without a full rebuild — is the job of attribute synchronization techniques, which depends on these same idempotent, checkpointed boundaries.

Operational Checklist

Spatial graph construction from OSM is not a one-time ETL job; it is a continuous topology-reconciliation process. With deterministic snapping, bounded concurrency, routing-ready edge weighting, and checkpointed partition boundaries, a team can hold sub-millisecond query latency while a metropolitan network grows into millions of nodes and edges.

This guide is part of Python for Spatial Graph Databases & Network Routing.

Subtopics

Siblings