Neo4j Import Skill

Neo4j Import Skill

When to Use

• Importing CSV, JSON, or Parquet files into Neo4j
• Batch-upserting nodes and relationships (UNWIND + CALL IN TRANSACTIONS)
• Migrating relational data (SQL → graph)
• Bulk-loading large datasets offline (neo4j-admin import)
• Choosing between online (Cypher) and offline (admin) import methods
• Verifying import completeness (counts, constraints, index states)

When NOT to Use

• Unstructured docs, PDFs, vector chunks → neo4j-document-import-skill
• Live application writes (MERGE/CREATE in app code) → neo4j-cypher-skill
• neo4j-admin backup/restore/config → neo4j-cli-tools-skill
• GDS algorithm projection from existing graph → neo4j-gds-skill

---

Method Decision Table

Aura: only https:// URLs — no file:///. Use neo4j-admin import only on self-managed.

---

Pre-Import Checklist

Run in this exact order — skipping causes hard-to-debug duplicates or missed index usage:

Constraints BEFORE import. Additional indexes AFTER import.

• Constraints create implicit RANGE indexes used by MERGE during load + enforce uniqueness
• Additional non-unique indexes (TEXT, RANGE on non-key props, FULLTEXT) created after load — Neo4j populates them async from the committed data; poll populationPercent until 100%
• Creating extra indexes before import slows every write during load with no benefit

1. Create uniqueness constraints (enables index used by MERGE):

   CREATE CONSTRAINT IF NOT EXISTS FOR (n:Person) REQUIRE n.id IS UNIQUE;
   CREATE CONSTRAINT IF NOT EXISTS FOR (n:Movie)  REQUIRE n.movieId IS UNIQUE;

Neo4j 2026.02+ (Enterprise/Aura) — PREVIEW: ALTER CURRENT GRAPH TYPE SET { … } can replace all individual constraint statements with a single declarative block. See neo4j-cypher-skill/references/graph-type.md. Use individual CREATE CONSTRAINT on older versions or Community Edition.

2. *Verify APOC if using apoc. procedures**:

   RETURN apoc.version();

If fails → APOC not installed. Use plain LOAD CSV instead.

3. Confirm target is PRIMARY (not replica):

   CALL dbms.cluster.role() YIELD role RETURN role;

If role ≠ PRIMARY → stop. Redirect write to PRIMARY endpoint.

4. Count source file rows before import (catch encoding issues early):

   wc -l data/persons.csv    # Linux/macOS

5. Verify UTF-8 encoding — LOAD CSV requires UTF-8. Re-encode if needed:

   file -i persons.csv       # Check encoding
   iconv -f latin1 -t utf-8 persons.csv > persons_utf8.csv

---

LOAD CSV Patterns

Basic node import with type coercion and null handling

CYPHER 25
LOAD CSV WITH HEADERS FROM 'file:///persons.csv' AS row
CALL (row) {
  MERGE (p:Person {id: row.id})
  ON CREATE SET
    p.name       = row.name,
    p.age        = toIntegerOrNull(row.age),
    p.score      = toFloatOrNull(row.score),
    p.active     = toBoolean(row.active),
    p.born       = CASE WHEN row.born IS NOT NULL AND row.born <> '' THEN date(row.born) ELSE null END,
    p.createdAt  = datetime()
  ON MATCH SET
    p.updatedAt  = datetime()
} IN TRANSACTIONS OF 10000 ROWS
  ON ERROR CONTINUE
  REPORT STATUS AS s
RETURN s.transactionId, s.committed, s.errorMessage

Null/empty-string rules:

• CSV missing column → null (safe)
• CSV empty string "" → stored as "" not null — use nullIf(row.x, '') to convert
• toInteger(null) throws → always use toIntegerOrNull()
• toFloat(null) throws → always use toFloatOrNull()
• Neo4j never stores null properties — they are silently dropped on SET

Relationship import (nodes must exist first)

CYPHER 25
LOAD CSV WITH HEADERS FROM 'file:///knows.csv' AS row
CALL (row) {
  MATCH (a:Person {id: row.fromId})
  MATCH (b:Person {id: row.toId})
  MERGE (a)-[:KNOWS {since: toIntegerOrNull(row.year)}]->(b)
} IN TRANSACTIONS OF 5000 ROWS
  ON ERROR CONTINUE
  REPORT STATUS AS s

Always import ALL nodes before ANY relationships — MATCH fails on missing nodes.

Tab-separated or custom delimiter

CYPHER 25
LOAD CSV WITH HEADERS FROM 'file:///data.tsv' AS row FIELDTERMINATOR '\t'
CALL (row) { MERGE (p:Person {id: row.id}) }
IN TRANSACTIONS OF 10000 ROWS ON ERROR CONTINUE

Compressed files (ZIP / gzip — local files only)

LOAD CSV WITH HEADERS FROM 'file:///archive.csv.gz' AS row ...

Cloud storage (Enterprise Edition)

Useful built-in functions inside LOAD CSV

linenumber()   // current line number — use as fallback ID
file()         // absolute path of file being loaded

---

CALL IN TRANSACTIONS — Full Reference

Syntax

CALL (row) {
  // write logic
} IN [n CONCURRENT] TRANSACTIONS
  [OF batchSize ROW[S]]
  [ON ERROR {CONTINUE | BREAK | FAIL | RETRY [FOR duration SECONDS] [THEN {CONTINUE|BREAK|FAIL}]}]
  [REPORT STATUS AS statusVar]

ON ERROR modes

ON ERROR CONTINUE/BREAK → outer transaction succeeds even if inner batches fail. ON ERROR FAIL → cannot be combined with REPORT STATUS AS.

CONCURRENT TRANSACTIONS (parallel batches)

CYPHER 25
LOAD CSV WITH HEADERS FROM 'file:///large.csv' AS row
CALL (row) {
  MERGE (p:Person {id: row.id}) SET p.name = row.name
} IN 4 CONCURRENT TRANSACTIONS OF 5000 ROWS
  ON ERROR RETRY FOR 30 SECONDS THEN CONTINUE
  REPORT STATUS AS s

Use CONCURRENT for read-heavy MERGE on non-overlapping key spaces. Risk: deadlocks on overlapping writes → combine with ON ERROR RETRY.

REPORT STATUS columns

Batch size guidance

---

neo4j-admin import (Offline Bulk Load)

Fastest method: ~3 min for 31M nodes / 78M rels on SSD. DB must be stopped or non-existent.

Command structure

neo4j-admin database import full \
  --nodes=Person="persons_header.csv,persons.csv" \
  --nodes=Movie="movies_header.csv,movies.csv" \
  --relationships=ACTED_IN="acted_in_header.csv,acted_in.csv" \
  --relationships=DIRECTED="directed_header.csv,directed.csv" \
  --delimiter=, \
  --id-type=STRING \
  --bad-tolerance=0 \
  --threads=$(nproc) \
  --high-parallel-io=on \
  neo4j

For SSDs: always set --high-parallel-io=on. For large graphs (>34B nodes/rels): --format=block.

Dry run (2026.02+) — validate without writing:

neo4j-admin database import full --dry-run ...

Node header file format

# persons_header.csv
personId:ID,name,born:int,score:float,active:boolean,:LABEL

# persons.csv (data file — no header row)
p001,Alice,1985,9.2,true,Person
p002,Bob,1990,7.1,false,Person

Relationship header file format

# acted_in_header.csv
:START_ID(Person),:END_ID(Movie),role,:TYPE

# acted_in.csv
p001,tt0133093,Neo,ACTED_IN
p002,tt0133093,Morpheus,ACTED_IN

:START_ID / :END_ID must reference the same :ID group as the node files.

Key flags

Schema file (--schema) [Enterprise, block format]

Pass a Cypher file with CREATE CONSTRAINT / CREATE INDEX statements; executed automatically after import completes. Constraints are created first (correct order enforced). File paths can be local or remote (s3://, gs://, https://).

neo4j-admin database import full \
  --format=block \
  --schema=schema.cypher \
  --nodes=Person="persons_header.csv,persons.csv" \
  neo4j

// schema.cypher
CREATE CONSTRAINT person_id IF NOT EXISTS FOR (n:Person) REQUIRE n.id IS UNIQUE;
CREATE CONSTRAINT movie_id  IF NOT EXISTS FOR (n:Movie)  REQUIRE n.id IS UNIQUE;
CREATE RANGE INDEX person_email IF NOT EXISTS FOR (n:Person) ON (n.email);
CREATE TEXT  INDEX movie_title  IF NOT EXISTS FOR (n:Movie)  ON (n.title);

For incremental import, DROP CONSTRAINT / DROP INDEX are also supported [2025.02+] — used to remove indexes before the merge phase and recreate them after for faster writes.

---

Incremental import (Enterprise only)

Three-phase process — use when DB must stay online during import preparation:

# Phase 1: Prepare staging area
neo4j-admin database import incremental --stage=prepare \
  --nodes=Person=persons_header.csv,delta.csv --force neo4j

# Phase 2: Build indexes (DB can be read-only during this phase)
neo4j-admin database import incremental --stage=build neo4j

# Phase 3: Merge into live database (brief write-lock)
neo4j-admin database import incremental --stage=merge neo4j

Requires Enterprise Edition + block store format.

---

APOC Patterns (when APOC is available)

Verify first: RETURN apoc.version() — if fails, use LOAD CSV or driver instead.

apoc.periodic.iterate — batch-process existing graph data

CALL apoc.periodic.iterate(
  "MATCH (p:Person) WHERE NOT (p)-[:HAS_ACCOUNT]->() RETURN p",
  "CREATE (p)-[:HAS_ACCOUNT]->(a:Account {id: randomUUID()})",
  {batchSize: 10000, parallel: false, retries: 2}
) YIELD batches, total, errorMessages
RETURN batches, total, errorMessages

Prefer CALL IN TRANSACTIONS (native Cypher) over apoc.periodic.iterate for new code — it has REPORT STATUS, CONCURRENT, and RETRY built in without APOC dependency.

apoc.load.csv — load with config options

CALL apoc.load.csv('file:///persons.csv', {
  header: true,
  sep: ',',
  skip: 1,
  limit: 1000000
}) YIELD lineNo, map, list
CALL (map) {
  MERGE (p:Person {id: map.id}) SET p.name = map.name
} IN TRANSACTIONS OF 10000 ROWS ON ERROR CONTINUE

apoc.load.json — load JSON from file or URL

CALL apoc.load.json('https://api.example.com/persons') YIELD value
CALL (value) {
  MERGE (p:Person {id: value.id}) SET p.name = value.name
} IN TRANSACTIONS OF 1000 ROWS ON ERROR CONTINUE

---

Driver Batch Write Pattern

Use when source is not a file (API responses, DB migrations). Collect into BATCH_SIZE (10 000) lists, call UNWIND $rows AS row MERGE ... per batch. ~10x faster than row-at-a-time. → Python + JS examples

---

MCP Tool Usage

Write gate — before any bulk write via MCP, show: 1. Query + affected labels 2. Estimated row count from source 3. EXPLAIN plan

Wait for user confirmation. Never auto-execute CALL IN TRANSACTIONS or CREATE CONSTRAINT without confirmation.

Always pass database param if not default: {"code": "...", "database": "neo4j"}.

---

Common Errors

---

Post-Import Validation

After import completes — run all:

// Row counts per label
MATCH (n:Person) RETURN count(n) AS persons;
MATCH ()-[:KNOWS]->() RETURN count(*) AS knows_rels;

// After import: create additional non-unique indexes (populated async)
CREATE TEXT INDEX movie_title IF NOT EXISTS FOR (n:Movie) ON (n.title);
CREATE RANGE INDEX person_born IF NOT EXISTS FOR (n:Person) ON (n.born);

// Poll population — wait until populationPercent = 100 before opening to queries
SHOW INDEXES YIELD name, state, populationPercent
WHERE state <> 'ONLINE' OR populationPercent < 100
RETURN name, state, populationPercent
ORDER BY populationPercent;

// Spot check: null keys = import bug
MATCH (p:Person) WHERE p.id IS NULL RETURN count(p) AS missing_id;

Do NOT run production queries until all indexes are ONLINE.

---

References

• LOAD CSV — Cypher Manual 25
• CALL IN TRANSACTIONS — Cypher Manual
• neo4j-admin database import
• APOC periodic execution
• APOC load procedures
• GraphAcademy: Importing CSV Data
• Indexes and constraints — types, MERGE lock semantics, import pre-flight
• Data Importer GUI — when to use, Aura access, multi-pass, gotchas
• Post-import refactoring — split lists, extract nodes, add labels, FK validation

---

Checklist

• [ ] Uniqueness constraints created before any MERGE-based import
• [ ] APOC availability verified if using apoc.* procedures
• [ ] Target confirmed as PRIMARY (not replica)
• [ ] Source files validated: UTF-8 encoding, expected row count, no BOM
• [ ] LOAD CSV uses toIntegerOrNull() / toFloatOrNull() — never bare toInteger()/toFloat()
• [ ] nullIf(row.x, '') applied where empty string ≠ null
• [ ] CALL IN TRANSACTIONS used (not USING PERIODIC COMMIT)
• [ ] ON ERROR CONTINUE + REPORT STATUS for production loads
• [ ] Node import completed before relationship import
• [ ] neo4j-admin: --bad-tolerance=0 set; --high-parallel-io=on for SSD
• [ ] Post-import: row counts match source; all indexes ONLINE
• [ ] Write execution gate applied (MCP): showed query + estimate, got confirmation
• [ ] Credentials in .env; .env in .gitignore

Status: Draft / WIP

neo4j-import-skill

Guides agents through importing structured data into Neo4j: LOAD CSV, batch upserts with CALL IN TRANSACTIONS, and offline bulk load with neo4j-admin import.

Install:

npx skills add https://github.com/neo4j-contrib/neo4j-skills --skill neo4j-import-skill

Or paste this link into your coding assistant: https://github.com/neo4j-contrib/neo4j-skills/tree/main/neo4j-import-skill

Neo4j Data Importer GUI

When to Use

Access in Aura

1. Log in: console.neo4j.io 2. Open AuraDB instance 3. Click Import in left sidebar — Data Importer opens pre-connected

Standalone URL (any Neo4j version): https://data-importer.neo4j.io/versions/0.7.0/?acceptTerms=true Provide WebSocket Bolt URL + password to connect.

Requirements

• CSV files on local filesystem (drag-and-drop into Files pane)
• CSV must have headers
• CSV must be clean (no encoding errors, consistent delimiters)
• IDs must be unique per node type
• DBMS must be running

Import Steps

1. Upload CSV(s) to Files pane (drag or Browse) 2. Click Add node label — enter label, select CSV file, map columns 3. Set unique ID (key icon) — Data Importer auto-creates uniqueness constraint + index 4. Drag edge between nodes to create relationship — select type, CSV file, from/to ID columns 5. Add optional relationship properties 6. Click Run import 7. View summary; verify in Query tool

Data Types Supported

Data Importer stores: String, Integer (Long), Float (Double), Boolean, Datetime.

Lists/arrays NOT supported — stored as delimited strings. Post-process with split().

What Data Importer Creates Automatically

• Uniqueness constraint on each node's unique ID property
• Index for each constrained property
• MERGE semantics on re-import (no duplicates if re-run)

Multi-pass for De-normalized Data

De-normalized CSV (one row = person + movie + role) requires multiple passes:

• Cannot create multiple node types from one file in single pass via GUI
• Pass 1: Map CSV → Person nodes; import
• Pass 2: Map same CSV → Movie nodes; import
• Pass 3: Map same CSV → ACTED_IN relationships; import

Model Save / Export

• Save model: give it a name, click Save (auto-saved on change)
• Export: ... menu → Download model (with data) — ZIP with mappings + CSVs
• Restore: ... menu → Open model (with data)

Common Mistakes

No unique ID set: duplicate nodes on re-import; relationship creation fails. Set key icon on ID column before running.

Foreign key kept as property: e.g. order_id as property instead of relationship. Foreign keys → relationships.

Type mismatch (silent failure): if Data Importer can't convert a value to the specified type, import succeeds but property is silently omitted. Verify node counts + spot-check properties; use string import + Cypher coercion if needed.

All data imports as strings: set correct type per column in mapping panel, or post-process with toInteger(), date(), split().

Importing before constraint creation: Data Importer creates constraints automatically (GUI path only). For Cypher path: create constraints manually BEFORE import.

Driver Batch Write Pattern (Python)

Use when source is not a file: API responses, database migrations, programmatic generation.

from neo4j import GraphDatabase

driver = GraphDatabase.driver("neo4j+s://xxx.databases.neo4j.io",
                              auth=("neo4j", "password"))

BATCH_SIZE = 10_000

def import_batch(tx, rows):
    tx.run("""
        UNWIND $rows AS row
        MERGE (p:Person {id: row.id})
        ON CREATE SET p.name = row.name, p.age = row.age
    """, rows=rows)

all_rows = [...]  # your source data

with driver.session(database="neo4j") as session:
    batch = []
    for row in all_rows:
        batch.append(row)
        if len(batch) == BATCH_SIZE:
            session.execute_write(import_batch, batch)
            batch.clear()
    if batch:
        session.execute_write(import_batch, batch)

driver.close()

UNWIND-based batching: ~10x faster than one-at-a-time — network round-trips are the bottleneck.

JavaScript / Node.js

const neo4j = require('neo4j-driver');
const driver = neo4j.driver('neo4j+s://xxx.databases.neo4j.io',
  neo4j.auth.basic('neo4j', 'password'));

const BATCH_SIZE = 10_000;
const session = driver.session({ database: 'neo4j' });

const rows = [...]; // your source data
for (let i = 0; i < rows.length; i += BATCH_SIZE) {
  const batch = rows.slice(i, i + BATCH_SIZE);
  await session.executeWrite(tx =>
    tx.run('UNWIND $rows AS row MERGE (p:Person {id: row.id}) SET p += row',
           { rows: batch })
  );
}
await session.close();
await driver.close();

Post-Import Refactoring Patterns

Use after Data Importer or basic LOAD CSV to reshape imported data.

1. Verify Property Types First

// Node property types (requires APOC)
CALL apoc.meta.nodeTypeProperties()
YIELD nodeType, propertyName, propertyTypes
RETURN nodeType, propertyName, propertyTypes
ORDER BY nodeType, propertyName;

// Relationship property types (requires APOC)
CALL apoc.meta.relTypeProperties()
YIELD relType, propertyName, propertyTypes
RETURN relType, propertyName, propertyTypes;

Without APOC: extract a sample and inspect values manually.

Neo4j Browser displays dates as strings — use n.born.year to confirm a property is a date, not a string.

2. String → List (split delimited field)

Data Importer stores multi-value fields as strings like "USA|Germany|France". Convert to StringArray:

MATCH (m:Movie)
CALL (m) {
  SET m.countries = split(coalesce(m.countries, ''), '|'),
      m.languages = split(coalesce(m.languages, ''), '|')
} IN TRANSACTIONS OF 10000 ROWS ON ERROR CONTINUE;

coalesce(m.countries, '') → returns '' if null; split('', '|') → [''] — filter afterward if needed:

SET m.countries = [x IN split(coalesce(m.countries, ''), '|') WHERE x <> '']

Check source data for separator (|, ,, ; are common). Never assume , — conflicts with CSV delimiter.

3. Add Labels Based on Relationships

Add specific labels for targeted lookups.

// Add Actor label to Person nodes with ACTED_IN relationship
MATCH (p:Person)-[:ACTED_IN]->()
SET p:Actor;

// Add Director label to Person nodes with DIRECTED relationship
MATCH (p:Person)-[:DIRECTED]->()
SET p:Director;

After adding, verify:

MATCH (n:Actor) RETURN count(n);
MATCH (n:Director) RETURN count(n);

4. Extract Nodes from String/List Property

Convert a property that holds category values into proper nodes + relationships.

Step 1: Create constraint for new node type

CREATE CONSTRAINT genre_name IF NOT EXISTS
FOR (g:Genre) REQUIRE g.name IS UNIQUE;

Step 2: UNWIND list → MERGE nodes + relationships

MATCH (m:Movie)
WHERE m.genres IS NOT NULL
CALL (m) {
  UNWIND m.genres AS genreName
  MERGE (g:Genre {name: genreName})
  MERGE (m)-[:IN_GENRE]->(g)
} IN TRANSACTIONS OF 10000 ROWS ON ERROR CONTINUE;

Step 3: Remove the now-redundant property

MATCH (m:Movie) WHERE m.genres IS NOT NULL
CALL (m) { REMOVE m.genres }
IN TRANSACTIONS OF 10000 ROWS ON ERROR CONTINUE;

Step 4: Verify schema

CALL db.schema.visualization();

5. String → Date/Datetime

MATCH (p:Person)
WHERE p.born IS NOT NULL
CALL (p) {
  SET p.born = date(p.born)
} IN TRANSACTIONS OF 10000 ROWS ON ERROR CONTINUE;

Ensure source string is ISO format (YYYY-MM-DD). For other formats, transform in the SET clause:

SET p.born = date({year: toInteger(left(p.born, 4)),
                   month: toInteger(substring(p.born, 5, 2)),
                   day: toInteger(right(p.born, 2))})

6. Validate Foreign Keys Before Creating Relationships

Detect missing referenced nodes before bulk relationship creation (prevents silent skips):

// Check for order rows where the customer doesn't exist yet
LOAD CSV WITH HEADERS FROM 'file:///orders.csv' AS row
WITH row.customer_id AS custId
WHERE NOT EXISTS { MATCH (c:Customer {customerID: custId}) }
RETURN DISTINCT custId AS missingCustomer
LIMIT 25;

Any results → import missing nodes first, then create relationships.

7. Self-Referencing Relationship (two-pass)

Node must exist before relationship can reference it — even when both ends share a label.

// Pass 1: Import Employee nodes (already done)

// Pass 2: Create REPORTS_TO from same file
LOAD CSV WITH HEADERS FROM 'file:///employees.csv' AS row
WHERE row.reports_to IS NOT NULL
CALL (row) {
  MATCH (e:Employee {employeeID: toIntegerOrNull(row.employee_id)})
  MATCH (m:Employee {employeeID: toIntegerOrNull(row.reports_to)})
  MERGE (e)-[:REPORTS_TO]->(m)
} IN TRANSACTIONS OF 5000 ROWS ON ERROR CONTINUE REPORT STATUS AS s;

Refactoring Checklist

• [ ] apoc.meta.nodeTypeProperties() run — types confirmed match data model
• [ ] Delimited string properties split into lists (split() + coalesce())
• [ ] Additional labels added (SET n:NewLabel) for query-targeted nodes
• [ ] Constraint created for new node type BEFORE extracting nodes
• [ ] Properties extracted to nodes via UNWIND + MERGE
• [ ] Source property removed after node extraction
• [ ] Self-referencing rels created in second pass (all nodes loaded first)
• [ ] FK validation run before any relationship creation pass
• [ ] Schema confirmed with CALL db.schema.visualization()