Async Batch Processing for Graphs
Spatial network routing demands deterministic latency and high-throughput topology ingestion. Raw OpenStreetMap extracts rarely arrive as clean, traversable structures. You must transform, validate, and persist millions of edges without blocking the main event loop. Async batch processing for graphs solves this architectural bottleneck by decoupling I/O from CPU-bound topology resolution. It enables continuous ingestion while maintaining strict memory ceilings and predictable query planner behavior.
When building foundational mobility networks, the transition from raw geospatial feeds to a query-ready topology is the most critical phase of Spatial Graph Construction & OSM Ingestion. Traditional synchronous loaders choke on connection pool exhaustion and transaction log bloat, especially when processing continental-scale PBF dumps. An async architecture replaces blocking database round-trips with cooperative multitasking. You stream coordinate arrays and edge attributes into bounded buffers. Each buffer feeds a coroutine queue that handles spatial snapping, projection normalization, and Cypher payload generation. The event loop schedules writes only when the connection pool signals readiness. For deeper architectural patterns on managing backpressure and worker lifecycle, consult Scaling Async Graph Ingestion with Python Asyncio.
Modern Python provides robust primitives for this workload. Use asyncio.Queue to bound concurrency and asyncio.Semaphore to cap active transactions. The official Neo4j async driver handles connection multiplexing natively, but you must avoid instantiating new connections per batch. Initialize a single driver instance at application startup and reuse it across all worker coroutines.
flowchart LR
Src["Source stream<br/>raw edges"] --> Val["Validate & snap<br/>haversine, bounds"]
Val --> Q[("Bounded asyncio.Queue<br/>maxsize = backpressure")]
Q --> W1["Worker 1"]
Q --> W2["Worker 2"]
Q --> Wn["Worker N"]
W1 & W2 & Wn --> Sem{{"Semaphore<br/>caps active tx"}}
Sem --> DB[("Graph database<br/>UNWIND batch")]
classDef in fill:#fbfaf6,stroke:#cdc6b3,color:#455062;
classDef q fill:#f6f0e6,stroke:#b58900,color:#5b3a0d;
classDef w fill:#e9f5f4,stroke:#0e7c86,color:#0e7c86;
classDef gate fill:#f5eef9,stroke:#5b21b6,color:#5b21b6;
classDef db fill:#fff,stroke:#c2410c,color:#c2410c;
class Src,Val in
class Q q
class W1,W2,Wn w
class Sem gate
class DB db
import asyncio
import math
from typing import List, Dict, Any
from neo4j import AsyncGraphDatabase, AsyncSession
class AsyncGraphBatchLoader:
def __init__(
self,
uri: str,
auth: tuple[str, str],
max_concurrency: int = 40,
batch_size: int = 5000
):
self.driver = AsyncGraphDatabase.driver(
uri, auth=auth, max_connection_pool_size=max_concurrency
)
self.semaphore = asyncio.Semaphore(max_concurrency)
self.queue: asyncio.Queue[List[Dict[str, Any]]] = asyncio.Queue(maxsize=2000)
self.batch_size = batch_size
@staticmethod
def _validate_and_snap(coords: tuple[float, float], precision: int = 6) -> tuple[float, float]:
"""Enforce WGS84 bounds and apply spatial snapping to prevent duplicate node creation."""
lat, lon = coords
if not (-90.0 <= lat <= 90.0 and -180.0 <= lon <= 180.0):
raise ValueError(f"Coordinates out of WGS84 bounds: {coords}")
return round(lat, precision), round(lon, precision)
@staticmethod
def _haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
"""Compute great-circle distance in meters for edge weight validation."""
R = 6371000.0
phi1, phi2 = math.radians(lat1), math.radians(lat2)
d_phi = math.radians(lat2 - lat1)
d_lambda = math.radians(lon2 - lon1)
a = math.sin(d_phi/2)**2 + math.cos(phi1) * math.cos(phi2) * math.sin(d_lambda/2)**2
return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
async def _execute_batch(self, tx: AsyncSession, chunk: List[Dict[str, Any]]) -> None:
query = """
UNWIND $batch AS row
MERGE (src:Node {id: row.src_id})
SET src.location = point({latitude: row.src_lat, longitude: row.src_lon}),
src.type = row.src_type
MERGE (tgt:Node {id: row.tgt_id})
SET tgt.location = point({latitude: row.tgt_lat, longitude: row.tgt_lon}),
tgt.type = row.tgt_type
MERGE (src)-[r:CONNECTS {id: row.edge_id}]->(tgt)
SET r.distance_m = row.distance_m,
r.travel_time_s = row.travel_time_s,
r.osm_tags = row.tags
"""
await tx.run(query, batch=chunk)
async def worker(self) -> None:
async with self.driver.session(database="neo4j") as session:
while True:
chunk = await self.queue.get()
try:
# execute_write handles automatic retries for transient errors
await session.execute_write(self._execute_batch, chunk)
except Exception as e:
# In production: route to dead-letter queue or implement exponential backoff
print(f"Batch ingestion failed: {e}")
finally:
self.queue.task_done()
async def ingest_stream(self, raw_edges: List[Dict[str, Any]]) -> None:
# Spawn fixed pool of worker coroutines
workers = [asyncio.create_task(self.worker()) for _ in range(5)]
# Stream chunks into bounded queue
for i in range(0, len(raw_edges), self.batch_size):
batch = raw_edges[i:i + self.batch_size]
validated_batch = []
for edge in batch:
s_lat, s_lon = self._validate_and_snap((edge['src_lat'], edge['src_lon']))
t_lat, t_lon = self._validate_and_snap((edge['tgt_lat'], edge['tgt_lon']))
# Validate spatial consistency before enqueueing
dist = self._haversine_distance(s_lat, s_lon, t_lat, t_lon)
validated_batch.append({
**edge,
'src_lat': s_lat, 'src_lon': s_lon,
'tgt_lat': t_lat, 'tgt_lon': t_lon,
'distance_m': round(dist, 2)
})
await self.queue.put(validated_batch)
await self.queue.join()
for w in workers:
w.cancel()Graph databases optimize for small, frequent transactions. Sending half a million edges in a single statement will exhaust heap memory and trigger aggressive garbage collection pauses. Split payloads into 5,000–10,000 record batches. Use UNWIND with parameterized maps to avoid query plan recompilation. Always pair MERGE operations with unique constraints on node identifiers. Without them, the engine falls back to full label scans, destroying throughput. Spatial indexes on point() properties must exist before ingestion begins. When enriching nodes with downstream attributes, coordinate alignment becomes critical. Misaligned geometries during POI Enrichment Workflows will break routing topology and invalidate spatial joins.
Async batch processing shifts bottlenecks from network latency to CPU scheduling and memory allocation. Large UNWIND payloads increase heap pressure, while micro-batches amplify transaction overhead and context switching. The optimal chunk size depends on your node degree distribution and available RAM. High-degree intersections (e.g., roundabouts, highway interchanges) require careful deduplication to prevent lock contention. Monitor the database’s transaction log growth and adjust heap allocation accordingly. For ingestion pipelines handling multi-terabyte PBF dumps, backpressure mechanisms must be implemented at the queue boundary to prevent OOM crashes. This aligns with robust OSM Data Ingestion Pipelines that prioritize deterministic throughput over raw ingestion speed.
By leveraging cooperative multitasking, bounded concurrency, and spatial validation gates, engineering teams can sustain high-throughput graph construction without sacrificing routing accuracy or system stability.