Graph Data Modeling

Automated Data Migration from Relational & JSON Sources

Migrating heterogeneous data estates into Neo4j 5.x requires a deterministic, production-grade automation framework rather than one-off ETL scripts. Ad-hoc pipelines fail under scale: they introduce duplicate nodes, orphaned relationships, and unpredictable transaction rollbacks that are expensive to unwind after cutover. This reference is written for platform teams, data modelers, and Python engineers who must move data out of normalized relational schemas and semi-structured JSON document stores and land it as a consistent, traversal-optimized property graph. It sets out the full architecture — source decomposition, mapping, chunked loading, contract validation, resilient execution, and operational readiness — and links each stage to the detailed guide that implements it. The recurring discipline throughout is explicit data contracts, stateless execution, and rigorous transaction lifecycle management so that a rerun of any stage produces exactly the same graph.

The pipeline below shows the end-to-end flow from heterogeneous sources into Neo4j.

End-to-end migration pipeline from relational and JSON sources into Neo4j A relational database and a JSON store both feed an Extract stage, which flows into Flatten and Map and then a Validate Contract gate. Records that pass the contract move to a Chunked Load and into Neo4j; records that fail are routed to a dead-letter queue. pass fail Relational DB JSON Store Extract Flatten & Map Validate Contract Chunked Load Neo4j target graph Dead-Letter Queue

Every stage below assumes a stable target topology. If you have not yet fixed your node and relationship model, design it first with the companion Neo4j Graph Schema Design & Architecture reference — a migration can only be as deterministic as the schema it targets.

Relational Schema Decomposition & Idempotent Mapping

Relational systems enforce rigid tabular structures that rarely align with graph topology. Direct row-to-node translation creates artificial junction nodes, inflates traversal depth, and degrades query performance. Production pipelines instead apply Relational Schema Mapping Strategies to decompose normalized tables into domain-aligned nodes and relationships: business entities become labeled nodes, foreign keys become explicit relationship types, and composite keys translate into NODE KEY constraints or deterministic synthetic identifiers. Getting the relationship semantics right here is a schema decision as much as a migration one, so lean on the sibling guidance for relationship cardinality and directionality when deciding edge direction and type.

The mapping layer must generate parameterized Cypher templates that guarantee idempotency. An idempotent MERGE with explicit ON CREATE and ON MATCH clauses prevents duplicate entity generation during concurrent or repeated execution. Always establish the backing constraint before ingestion so MERGE resolves against an index instead of scanning the label.

python
from neo4j import GraphDatabase

def ingest_customer(session, customer_id, name, region):
    # Neo4j 5.x parameterized MERGE with explicit lifecycle handling
    query = """
    MERGE (c:Customer {customer_id: $customer_id})
    ON CREATE SET c.name = $name, c.created_at = datetime()
    ON MATCH SET c.name = $name, c.last_synced = datetime()
    MERGE (r:Region {code: $region})
    MERGE (c)-[:LOCATED_IN]->(r)
    """
    session.execute_write(
        lambda tx: tx.run(query, customer_id=customer_id, name=name, region=region)
    )

Architectural trade-off: MERGE introduces locking overhead compared with CREATE. For initial loads exceeding 10M records, pre-create unique constraints, use CREATE with application-level deduplication during the bulk phase, and switch back to MERGE for ongoing delta syncs once the constraints are online. A concrete worked example of translating referential integrity is migrating PostgreSQL foreign keys to Neo4j relationships automatically.

Hierarchical JSON Normalization & Path Extraction

Document-oriented sources introduce structural volatility. Deeply nested arrays, polymorphic schemas, and inconsistent key naming must be normalized before graph ingestion. JSON Document Flattening & Graph Conversion establishes a transformation stage that extracts hierarchical paths into discrete graph entities while preserving referential lineage. Flattening operations should retain a source path (for example $.orders[0].items[2].sku) so every derived node is auditable and reverse-traceable back to the document that produced it.

python
import json
from collections import deque

def flatten_document(doc, root_path=""):
    """Iterative BFS flattening; avoids recursion stack overflow on deep payloads."""
    queue = deque([(doc, root_path)])
    result = []
    while queue:
        current, path = queue.popleft()
        if isinstance(current, dict):
            for key, value in current.items():
                new_path = f"{path}.{key}" if path else key
                if isinstance(value, (dict, list)):
                    queue.append((value, new_path))
                else:
                    result.append({"path": new_path, "value": value})
        elif isinstance(current, list):
            for idx, item in enumerate(current):
                item_path = f"{path}[{idx}]"
                if isinstance(item, (dict, list)):
                    queue.append((item, item_path))
                else:
                    result.append({"path": item_path, "value": item})
    return result

Map flattened paths to nodes using deterministic labels (Order, OrderItem) and connect them via structural relationships (:CONTAINS, :HAS_ATTRIBUTE). Enforce strict type coercion and explicit null handling, and avoid dynamic property keys in Cypher, which fragment the query planner’s statistics. Do not store raw JSON blobs as node properties: they bypass Neo4j’s native storage optimizations and inflate page-cache pressure — a classic case of the property graph anti-patterns the schema reference warns against. The hardest part in practice is repeated array structures, covered in handling nested JSON arrays during graph ingestion.

Driver Orchestration & Chunked Transaction Boundaries

High-volume ingestion demands controlled resource utilization and predictable memory footprints. The Python driver’s connection pooling and transaction management must be orchestrated through deterministic chunking boundaries. A disciplined batch-processing workflow defines the execution rhythm for parallelized data streams, where optimal batch sizing depends on relationship density, property payload size, and transaction log capacity (db.tx_log.rotation.retention_policy). Implement cursor-based pagination or watermark tracking so an interrupted pipeline resumes precisely at the last committed offset rather than replaying from the start.

python
from neo4j import AsyncGraphDatabase
from itertools import islice

def chunk_iter(iterable, size):
    it = iter(iterable)
    return iter(lambda: list(islice(it, size)), [])

async def run_chunked_ingestion(driver, records, batch_size=5000):
    async with driver.session(database="neo4j") as session:
        for chunk in chunk_iter(records, batch_size):
            await session.execute_write(
                lambda tx, c=chunk: tx.run(
                    "UNWIND $batch AS row MERGE (e:Entity {id: row.id}) SET e += row.props",
                    batch=c
                )
            )

Architectural trade-off: larger batches reduce network round-trips but increase transaction-log pressure and heap allocation. Start with 2,000–5,000 records per chunk, monitor neo4j-admin server report and JMX heap metrics, and reduce the batch size if GC pauses spike. For very large loads, switch to CALL { ... } IN TRANSACTIONS for server-side chunking — run it as an auto-commit query via session.run, never inside execute_write. When parallel workers MERGE the same key concurrently you can still create duplicates; the fix is detailed in resolving duplicate nodes during parallel batch loads.

Contract Enforcement & Data Integrity

Schema drift and silent data corruption undermine migration success. Data Validation & Integrity Checks mandate pre-ingestion contract validation and post-ingestion graph consistency audits. Enforce UNIQUE, NODE KEY, and relationship-property constraints at the database level before the pipeline runs, and validate data types, required properties, and relationship cardinality in the Python layer using Pydantic or Marshmallow before serialization. Because source systems evolve independently of the graph model, automated schema-drift detection should compare the live source contract against the graph’s expected shape and fail closed when they diverge.

python
from pydantic import BaseModel, field_validator

class CustomerRecord(BaseModel):
    customer_id: str
    name: str
    region: str

    @field_validator("customer_id")
    @classmethod
    def non_empty_key(cls, v: str) -> str:
        # Reject empty keys before they reach MERGE and create a null-keyed node
        if not v or not v.strip():
            raise ValueError("customer_id is required and must be non-empty")
        return v

A record that fails validation should never reach the graph; it is routed to a dead-letter queue for deterministic replay, keeping the load atomic at the contract boundary rather than trusting the database to reject malformed writes downstream.

Resilient Execution & Rollback Patterns

When transient failures occur, Error Handling & Rollback Mechanisms require explicit transaction boundaries and compensating actions. Neo4j’s ACID guarantees provide atomicity per transaction, but application-level retries must implement exponential backoff, jitter, and dead-lettering for malformed records. Never swallow Neo4jError exceptions: log the full stack trace, transaction metadata, and failed payload so the record can be replayed deterministically.

python
from neo4j.exceptions import Neo4jError
import time

def resilient_execute(session, query, params, max_retries=3):
    for attempt in range(max_retries):
        try:
            session.execute_write(lambda tx: tx.run(query, **params))
            return
        except Neo4jError as e:
            # Neo4j transient errors use the "Neo.TransientError.*" code prefix
            if e.code and e.code.startswith("Neo.TransientError"):
                time.sleep(2 ** attempt)
                continue
            raise
    raise RuntimeError("Max retries exceeded for transaction")

Because retries can re-execute a partially applied batch, every write in the pipeline must be idempotent by construction — the same discipline established in the mapping stage. This is why idempotent migration scripts are the precondition for safe retry logic rather than an afterthought.

Constraints & Index Lifecycle

The stages above share one prerequisite that belongs at the migration level rather than any single stage: the constraint and index topology must exist and be online before bulk writes begin. Constraints created up front turn every MERGE into an index seek and prevent duplicate-key contention; constraints created after a large CREATE load force a full backfill and can fail if the data already violates uniqueness. All DDL must be idempotent so a rerun is a no-op, which Neo4j 5.x expresses with IF NOT EXISTS.

cypher
// Run each statement separately over Bolt — one Cypher statement per call.
CREATE CONSTRAINT customer_key IF NOT EXISTS
FOR (c:Customer) REQUIRE c.customer_id IS UNIQUE;

CREATE CONSTRAINT order_key IF NOT EXISTS
FOR (o:Order) REQUIRE o.order_id IS NODE KEY;

// Range index to accelerate watermark/delta lookups during incremental sync
CREATE INDEX order_synced_at IF NOT EXISTS
FOR (o:Order) ON (o.last_synced);
Constraint and index lifecycle relative to the bulk load phase Two execution timelines. In the correct order, constraints and indexes are created and reach the ONLINE state before the bulk load, so every MERGE is an index seek and the rerun is deterministic with no duplicates. In the anti-pattern, the bulk CREATE load runs first with no keys, constraints are added afterward, and the resulting backfill scan fails on a duplicate key. Constraints first correct order CREATE CONSTRAINT + INDEX Wait until indexes ONLINE Bulk load MERGE = index seek No duplicates deterministic rerun Load first anti-pattern Bulk CREATE load (no keys) Add constraints after load Backfill scan over all rows Fails on duplicate key left to right = execution timeline

Provision constraints, wait for indexes to report ONLINE via SHOW INDEXES, and only then release the loaders. For the full staged sequence — heap sizing, page cache, and the order in which structural enforcement comes online — see Initial Load Performance Tuning.

Query Planner Implications

The mapping and constraint decisions made during migration are exactly what the Cypher planner relies on at query time. A uniqueness or NODE KEY constraint gives the planner a guaranteed single-row seek (NodeUniqueIndexSeek) as a traversal anchor; without it the same lookup degrades to a NodeByLabelScan whose cost grows with the label’s cardinality. Verify this with PROFILE on a representative query after a load:

cypher
PROFILE
MATCH (c:Customer {customer_id: $id})-[:LOCATED_IN]->(r:Region)
RETURN r.code;

In the profile output, the first operator should be NodeUniqueIndexSeek with db hits close to the number of rows returned. If you instead see NodeByLabelScan followed by a Filter, the backing constraint is missing or the property type does not match the index (a common outcome of loose JSON coercion). Because the planner caches statistics, run a load, then confirm the plan is stable — dynamic labels or dynamic property keys emitted by a sloppy flattening stage fragment those statistics and produce volatile plans. Aligning ingested types with deliberate graph data type selection keeps the planner’s cost model accurate.

End-to-End Driver Integration Pattern

The canonical migration workflow composes every stage into a single transactional session: validate the contract, chunk the stream, execute idempotent writes with retry, and record a watermark for resumability. The driver is instantiated once as a long-lived, thread-safe object and closed with a context manager; sessions are short-lived and scoped to a batch.

python
from neo4j import GraphDatabase
from neo4j.exceptions import Neo4jError
import time

UPSERT = """
UNWIND $batch AS row
MERGE (c:Customer {customer_id: row.customer_id})
ON CREATE SET c.created_at = datetime()
SET c.name = row.name, c.last_synced = datetime()
MERGE (r:Region {code: row.region})
MERGE (c)-[:LOCATED_IN]->(r)
"""

def load_batch(tx, batch):
    tx.run(UPSERT, batch=batch)

def migrate(uri, auth, batches):
    # One driver for the process lifetime; sessions are cheap and per-batch.
    with GraphDatabase.driver(uri, auth=auth) as driver:
        with driver.session(database="neo4j") as session:
            for chunk in batches:                       # pre-validated records
                for attempt in range(3):
                    try:
                        # execute_write wraps a managed transaction with
                        # automatic retry on the driver's own retryable errors
                        session.execute_write(load_batch, chunk)
                        break
                    except Neo4jError as e:
                        if e.code and e.code.startswith("Neo.TransientError") and attempt < 2:
                            time.sleep(2 ** attempt)
                            continue
                        raise  # non-transient: route chunk to the dead-letter queue

This pattern is deterministic end to end: because MERGE is idempotent and the watermark advances only on a committed batch, rerunning migrate after a crash converges on the same graph without duplicates. Consult the official Neo4j Python Driver Manual for the full transaction-function contract and connection-pool tuning.

Anti-Patterns & Failure Modes

These failure modes recur across relational and JSON migrations. Each has a clear diagnosis and a corrective action.

  1. Unconstrained MERGE (label scans and duplicates). Running MERGE before the backing constraint is online forces a full label scan per statement and lets concurrent workers create duplicate nodes. Diagnosis: PROFILE shows NodeByLabelScan; MATCH (n:Customer) WITH n.customer_id AS k, count(*) AS c WHERE c > 1 RETURN k, c returns rows. Fix: create the uniqueness/NODE KEY constraint and wait for ONLINE before any load.

  2. Oversized transactions (heap exhaustion). Committing an entire source table in one transaction inflates the transaction log and triggers OutOfMemoryError or long GC pauses. Diagnosis: GC pause spikes in JMX, growing tx_log on disk. Fix: chunk to 2,000–5,000 records, or use CALL { ... } IN TRANSACTIONS for server-side batching.

  3. Dynamic labels and property keys. Emitting labels or keys derived from JSON values (SET n[row.key] = ..., computed labels) explodes the token store and fragments planner statistics. Diagnosis: thousands of distinct labels in db.labels(); unstable query plans. Fix: map to a bounded node label taxonomy and store variance as properties.

  4. Raw JSON blobs as string properties. Persisting an entire document as a stringified property defeats indexing and bloats the page cache. Diagnosis: very large string properties; queries that CONTAINS-scan JSON text. Fix: flatten to discrete nodes and typed properties before ingestion.

  5. Swallowed errors and non-idempotent replay. Catching Neo4jError and continuing hides partial writes, and non-idempotent statements corrupt the graph on retry. Diagnosis: record counts drift between source and graph; duplicates after a restart. Fix: dead-letter failed records, make every write idempotent, and gate reruns on committed watermarks. Model changes over time with disciplined schema evolution and versioning so a replayed migration never fights an out-of-band schema change.

Production Readiness & Operational Considerations

Automated migration pipelines must integrate with observability stacks. Emit OpenTelemetry traces per chunk and track records_processed, failed_records, and throughput_rps as Prometheus-compatible metrics. Align ingestion windows with maintenance schedules so bulk writes do not contend with online analytical queries. Post-migration, use neo4j-admin database dump for logical backups and offline restore verification, and execute cutover only after dual-write validation and traffic shadowing confirm graph parity with the source system. For the tuning knobs that govern the heaviest phase, return to Initial Load Performance Tuning, and consult the Cypher Manual: Constraints and Indexes for exact DDL syntax.

Explore this section