Loading data directly from application code with UNWIND

Cypher®’s UNWIND is what lets an application send a whole batch of records to Neo4j in a single query, instead of one query per record.

The best practices for Schema and indexes, Optimized Cypher patterns for ingestion, and Transactional and concurrent Cypher ingestion apply to this page and are worth reviewing before continuing.

UNWIND as an online ingestion method

Rather than staging data as a CSV file for Neo4j to read, application code (using any Neo4j driver) can send rows directly, parameterized into an UNWIND query. This provides and online ingestion path, pushing data from wherever it already lives in an application, rather than exporting it to a file first.

Specifically, UNWIND turns a list of data from an application, passed in as a query parameter into individual rows that the rest of the query can MERGE/CREATE against one at a time. That matters because sending one query per row does not scale. For example, 100,000 separate MERGE statements means 100,000 round trips, each paying network latency even if the server processes each one instantly. Embedding values directly in the query text also stops Neo4j from reusing a cached plan, forcing a re-parse each time. UNWIND $rows avoids both: a single round trip can be made, with one cached plan.

Basic UNWIND exampl
UNWIND $rows AS row
MERGE (n:Label {id: row.id})
SET n += row.properties

Preparing and batching data client-side

UNWIND-based ingestion does not require any particular kind of source, only that application code, not Neo4j, is the one reading and preparing the data before sending it through a driver That data can come from:

  • A file the application reads itself, such as a CSV, parsed into a dataframe or a list of dictionaries before being batched and sent.

  • A queue consumer, receiving individual events (e.g. from Kafka, SQS) and accumulating them into batches before sending.

  • An API payload, where records arrive as part of a request the application is already handling.

  • An in-memory ETL job, transforming or joining data from other systems before writing the result to Neo4j.

For example, transfers.csv read by an application might look like this before loading:

transfers.csv
fromAccountId,toAccountId,transferId,amount,timestamp
101,202,9001,250.00,2026-08-11T09:15:00+00:00
305,101,9002,40.50,2026-08-11T09:16:30+00:00
202,305,9003,1200.00,2026-08-11T09:18:12+00:00

If the dataset fits in memory, load it into a dataframe and convert each column to the correct datatype before sending anything to Neo4j. Fixing types here is cheaper and easier than dealing with data exception errors caused by constraint violations once in Neo4j.

import pandas as pd

df = pd.read_csv("transfers.csv")
df["transferId"] = df["transferId"].astype(int)
df["amount"] = df["amount"].astype(float)
df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)

Split the data into batches, and send each batch as one UNWIND query rather than sending everything in a single call. Without this, one enormous parameter payload has to be built in memory, serialized, and sent in one go. Batching client-side also gives a natural checkpoint: if something fails partway through, it is clear exactly which chunks succeeded and which did not.

A batch size in the low thousands (start around 1,000–10,000 rows and adjust based on observed performance) keeps each transaction a manageable size.

batch_size = 5000

for start in range(0, len(df), batch_size):
    batch = df.iloc[start:start + batch_size].to_dict("records")
    driver.execute_query(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (from:Account {id: row.fromAccountId})
          MERGE (to:Account {id: row.toAccountId})
          MERGE (from)-[t:TRANSFER {id: row.transferId}]->(to)
          SET t.amount = row.amount, t.timestamp = row.timestamp
        } IN TRANSACTIONS OF 500 ROWS
        """,
        rows=batch,
        database_="neo4j",
    )
CALL { …​ } IN TRANSACTIONS requires auto-commit: it cannot run inside an explicit transaction. execute_query() runs as its own auto-commit unit, which is why it is used here instead of session.execute_write(), a transaction function that would wrap the query in exactly the kind of explicit transaction this construct is incompatible with.

batch (the rows parameter) looks like this, once converted:

[
    {"fromAccountId": 101, "toAccountId": 202, "transferId": 9001, "amount": 250.0, "timestamp": Timestamp("2026-08-11 09:15:00+0000")},
    {"fromAccountId": 305, "toAccountId": 101, "transferId": 9002, "amount": 40.5, "timestamp": Timestamp("2026-08-11 09:16:30+0000")},
    {"fromAccountId": 202, "toAccountId": 305, "transferId": 9003, "amount": 1200.0, "timestamp": Timestamp("2026-08-11 09:18:12+0000")},
]

Inside the Cypher, UNWIND $rows AS row turns this list into individual rows, one at a time, so row.fromAccountId, row.toAccountId, row.transferId, row.amount, and row.timestamp refer to that row’s specific values on each pass through the query. Note that batching happens at two levels here, not one: the Python loop above batches client-side (5,000 rows per request); CALL { …​ } IN TRANSACTIONS OF 500 ROWS then batches server-side within that request, so a 5,000-row request does not become one single 5,000-row transaction either.

Concurrent UNWIND ingestion

Sending every row as one large UNWIND statement risks running out of transactional memory once the dataset gets large. Wrapping UNWIND within a CALL { …​ } IN TRANSACTIONS splits the rows into smaller batches, each committed as its own transaction, so memory use stays bounded regardless of total dataset size. However, batches committed one after another are still limited by how fast a single transaction can run. Running batches concurrently, via IN CONCURRENT TRANSACTIONS, lets several batches write at the same time, reducing total ingestion time.

Wrapping UNWIND ingestion in a CALL {…​} IN CONCURRENT TRANSACTIONS subquery
def load_batch(driver, rows):
    driver.execute_query(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (from:Account {id: row.fromAccountId})
          MERGE (to:Account {id: row.toAccountId})
          MERGE (from)-[t:TRANSFER {id: row.transferId}]->(to)
          SET t.amount = row.amount, t.timestamp = row.timestamp
        } IN CONCURRENT TRANSACTIONS OF 500 ROWS
        """,
        rows=rows,
        database_="neo4j",
    )

for batch in chunks(all_transfers, size=5000):
    load_batch(driver, batch)

driver.close() print("Done")

Concurrency introduces a new risk: any two concurrent transactions that need to acquire locks on the same nodes or relationships, in a different order, can deadlock. For example, every TRANSFER relationship touches two Account nodes, so the same account can easily appear in more than one row, making a deadlock likely at scale. For more information, see Best practices → Concurrent transactions.

Example

Remember to create an index-backed schema before ingesting. For more information, see Best practices → Index-backed key constraints and type constraints
Example 1. Online ingestion using UNWIND
# load_transfers.py

import random
from datetime import datetime, timedelta, timezone
from neo4j import GraphDatabase

# Step 1: simulate the application data source
def fetch_transfer_batch_from_queue(n):
    base_time = datetime(2026, 8, 11, tzinfo=timezone.utc)
    return [
        {
            "fromAccountId": random.randint(1, 5000),
            "toAccountId": random.randint(1, 5000),
            "transferId": i,
            "amount": round(random.uniform(5, 2000), 2),
            "timestamp": base_time + timedelta(seconds=i),
        }
        for i in range(n)
    ]

all_transfers = fetch_transfer_batch_from_queue(200_000)

# Step 3: driver and batching setup
uri = "bolt://localhost:7687"
user = "<database>"
password = "<password>"

driver = GraphDatabase.driver(uri, auth=(user, password))
driver.verify_connectivity()

def chunks(data, size):
    for i in range(0, len(data), size):
        yield data[i:i + size]

# Step 4: ingest with concurrent transactions and DISJOINT BY
def load_batch(driver, rows):
    with driver.session(database="neo4j") as session:
        session.run(
            """
            UNWIND $rows AS row
            CALL (row) {
              MERGE (from:Account {id: row.fromAccountId})
              MERGE (to:Account {id: row.toAccountId})
              MERGE (from)-[t:TRANSFER {id: row.transferId}]->(to)
              SET t.amount = row.amount, t.timestamp = row.timestamp
            } IN CONCURRENT TRANSACTIONS OF 500 ROWS
              DISJOINT BY (row.fromAccountId, row.toAccountId)
              ON ERROR RETRY
            """,
            rows=rows,
        ).consume()

for batch in chunks(all_transfers, size=5000):
    load_batch(driver, batch)

records, _, _ = driver.execute_query(
      """
      RETURN COUNT { MATCH (a:Account) } AS accountCount,
             COUNT { MATCH ()-[t:TRANSFER]->() } AS transferCount
      """,
      database_="neo4j",
      )

print(f"Accounts: {records[0]['accountCount']}")
print(f"Transfers: {records[0]['transferCount']}")

driver.close()
print("Done")