neo4j-import-skill

Neo4j 导入技能

适用场景

• 将 CSV、JSON 或 Parquet 文件导入 Neo4j
• 批量插入或更新节点和关系（UNWIND + CALL IN TRANSACTIONS）
• 迁移关系型数据（SQL → 图）
• 离线批量加载大型数据集（neo4j-admin import）
• 在在线（Cypher）和离线（admin）导入方法之间进行选择
• 验证导入是否完整（计数、约束、索引状态）

不适用场景

• 非结构化文档、PDF、向量分块 → neo4j-document-import-skill
• 实时应用写入（应用代码中的 MERGE/CREATE） → neo4j-cypher-skill
• neo4j-admin 备份/恢复/配置 → neo4j-cli-tools-skill
• 从现有图投影 GDS 算法 → neo4j-gds-skill

---

方法决策表

数据集大小	数据库状态	来源	方法
任意大小	在线	CSV（Aura 或本地）	LOAD CSV + CALL IN TRANSACTIONS
< 1M 行	在线	列表/API 响应	UNWIND + CALL IN TRANSACTIONS
> 10M 行	离线（本地/自管理）	CSV / Parquet	neo4j-admin database import full
任意大小	在线	APOC 可用	apoc.periodic.iterate + apoc.load.csv
任意大小	在线	JSON/API	apoc.load.json 或驱动程序批处理
增量差异数据	离线（Enterprise）	CSV	neo4j-admin database import incremental

Aura：仅支持 https:// URL，不支持 file:///。仅在可自托管环境中使用 neo4j-admin import。

---

导入前检查清单

请严格按以下顺序执行——跳过步骤会导致难以排查的重复数据或索引未被使用：

导入前创建约束。导入后创建额外索引。

• 约束会创建隐式 RANGE 索引，供加载期间的 MERGE 使用，并确保唯一性
• 其他非唯一索引（TEXT、非键属性上的 RANGE、FULLTEXT）应在加载后创建——Neo4j 会根据已提交的数据异步填充这些索引；轮询 populationPercent，直到其达到 100%
• 导入前创建额外索引会拖慢加载期间的每次写入，且没有任何好处

1. 创建唯一性约束（启用 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.06+（Enterprise/Aura，GA）： ALTER CURRENT GRAPH TYPE SET { … } 可用单个声明式代码块替代所有单独的约束语句。请参阅 neo4j-cypher-skill/references/graph-type.md。Community Edition 或 2026.02 之前的版本请使用单独的 CREATE CONSTRAINT。

2. 如果使用 apoc. 过程，请验证 APOC*：

   RETURN apoc.version();
   

   如果失败 → 未安装 APOC。请改用普通 LOAD CSV。

3. 确认目标是 PRIMARY（而非副本）：

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

   如果角色 ≠ PRIMARY → 停止。将写入重定向到 PRIMARY 端点。

4. 导入前统计源文件行数（及早发现编码问题）：

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

5. 验证 UTF-8 编码——LOAD CSV 要求使用 UTF-8。必要时重新编码：

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

---

LOAD CSV 模式

基本节点导入：类型转换与空值处理

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

空值/空字符串规则：

• CSV 缺少列 → null（安全）
• CSV 空字符串 "" → 存储为 ""，而不是 null——使用 nullIf(row.x, '') 进行转换
• toInteger(null) 会报错 → 始终使用 toIntegerOrNull()
• toFloat(null) 会报错 → 始终使用 toFloatOrNull()
• Neo4j 从不存储 null 属性——执行 SET 时会静默丢弃这些属性

关系导入（节点必须先存在）

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

务必先导入所有节点，再导入任何关系——缺少节点时 MATCH 会失败。

制表符分隔或自定义分隔符

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

压缩文件（ZIP / gzip——仅限本地文件）

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

云存储（Enterprise Edition）

协议	示例
AWS S3	s3://my-bucket/data/persons.csv
Google Cloud Storage	gs://my-bucket/persons.csv
Azure Blob	azb://account/container/persons.csv

LOAD CSV 中的实用内置函数

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

---

CALL IN TRANSACTIONS——完整参考

语法

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 模式

模式	行为	适用场景
ON ERROR FAIL	默认。首次出错时回滚整个外层事务	严格的全有或全无式导入
ON ERROR CONTINUE	跳过失败的批次，继续处理剩余批次	需要弹性的大规模加载——通过 REPORT STATUS 跟踪错误
ON ERROR BREAK	首个批次失败后停止；保留已完成的工作	半严格模式：提前停止，保留成功的批次
ON ERROR RETRY	指数退避重试（默认 30 秒）+ 后备方案	存在死锁风险的并发写入

即使内部批次失败，ON ERROR CONTINUE/BREAK → 外层事务仍会成功。 ON ERROR FAIL → 不能与 REPORT STATUS AS 组合使用。

CONCURRENT TRANSACTIONS（并行批次）

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

对于键空间不重叠、以读取为主的 MERGE，请使用 CONCURRENT。风险：写入重叠可能导致死锁 → 与 ON ERROR RETRY 组合使用。

REPORT STATUS 列

列	类型	含义
s.started	BOOLEAN	批次事务已启动
s.committed	BOOLEAN	批次已成功提交
s.transactionId	STRING	事务 ID
s.errorMessage	STRING or null	批次失败时的错误详情

批次大小建议

行数	建议批次大小	备注
< 100k	10 000	默认值即可
100k – 1M	10 000 – 50 000	监控堆内存；速度快时可增大
1M – 10M	50 000 – 100 000	CPU 可用时启用 CONCURRENT
> 10M online	50 000	可考虑改用 neo4j-admin import
关系导入	5 000	调低——每个批次会执行 2 次 MATCH

---

neo4j-admin import（离线批量加载）

速度最快：在 SSD 上，加载 31M 个节点 / 78M 条关系约需 3 分钟。数据库必须已停止或尚不存在。

命令结构

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

试运行（2026.02+）——验证但不写入：

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

节点头文件格式

# 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

字段	含义
:ID	用于关系连接的唯一 ID（默认不作为属性存储）
:ID(Group)	限定范围的 ID 空间——节点类型共用 ID 时使用
:LABEL	一个或多个标签；以分号分隔：Person;Employee
prop:int	带类型的属性；类型：int long float double boolean byte short string
prop:date	时态类型：date localtime time localdatetime datetime duration；Parquet 的 INTERVAL 列会作为 DURATION 导入 [2026.07+]
prop:int[]	数组——单元格中的值以分号分隔：1;2;3
prop:vector	浮点向量（2025.10+）——CSV 中坐标以分号分隔；可直接从原生 Parquet 列表类型导入 [2026.06+]

关系头文件格式

# 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 必须引用与节点文件相同的 :ID 组。

关键标志

标志	默认值	备注
--delimiter	,	单字节 UTF-8 字符、TAB、\ID 或 U+XXXX；不允许换行符 [2026.07+]
--vector-delimiter	;	分隔 prop:vector 坐标；必须与 --delimiter 和 --quote 不同 [2026.06+ 起强制执行]
--id-type	STRING	STRING | INTEGER | ACTUAL
--bad-tolerance	-1（无限制，2025.12 起更改）	生产环境严格导入时设为 0
--threads	CPU 数量	在共享主机上显式设置
--max-off-heap-memory	RAM 的 90%	如果主机还运行其他服务，请调低
--high-parallel-io	off	SSD/NVMe 环境下设为 on
--format	standard	超过 34B 个节点/关系时使用 block
--overwrite-destination	false	数据库已存在时必须设置
--dry-run	false	2026.02+——验证但不写入

Schema 文件（--schema）[Enterprise，block 格式]

传入包含 CREATE CONSTRAINT / CREATE INDEX 语句的 Cypher 文件；导入完成后会自动执行。首先创建约束（强制按正确顺序执行）。文件路径可以是本地路径，也可以是远程路径（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);

对于增量导入，也支持 DROP CONSTRAINT / DROP INDEX [2025.02+] —— 可在合并阶段前删除索引，并在合并阶段后重新创建，以加快写入速度。--schema 也接受图类型 DDL：full 可用 ALTER CURRENT GRAPH TYPE SET {…} [2026.05+]，incremental 可用 ALTER CURRENT GRAPH TYPE ADD/DROP/ALTER {…} [2026.06+] —— 请参阅 neo4j-cypher-skill/references/graph-type.md。

---

增量导入（仅限 Enterprise）

三阶段流程——用于导入准备期间数据库必须保持在线的情况：

# 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

需要 Enterprise Edition 和 block 存储格式。

---

APOC 模式（APOC 可用时）

先验证：RETURN apoc.version()——如果失败，请改用 LOAD CSV 或驱动程序。

apoc.periodic.iterate——批量处理现有图数据

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

配置键	默认值	说明
batchSize	10000	每个内部事务处理的行数
parallel	false	对不重叠的写入启用；风险：死锁
retries	0	以 100ms 间隔重试失败的批次 N 次

新代码优先使用 CALL IN TRANSACTIONS（原生 Cypher），而不是 apoc.periodic.iterate——它内置了 REPORT STATUS、CONCURRENT 和 RETRY，无需依赖 APOC。

apoc.load.csv——使用配置选项加载

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——从文件或 URL 加载 JSON

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

---

驱动程序批量写入模式

当数据源不是文件时（API 响应、数据库迁移），请使用此模式。将数据收集到 BATCH_SIZE（10 000）大小的列表中，并针对每个批次调用 UNWIND $rows AS row MERGE ...。速度约为逐行处理的 10 倍。→ Python 和 JS 示例

---

MCP 工具用法

操作	MCP 工具	说明
SHOW CONSTRAINTS、SHOW INDEXES	read-cypher	导入前务必检查
CREATE CONSTRAINT、CREATE INDEX	write-cypher	审核步骤：展示计划创建的约束并确认
LOAD CSV / CALL IN TRANSACTIONS	write-cypher	审核步骤：展示行数和 Cypher，并确认
验证计数	read-cypher	导入后：MATCH (n:Label) RETURN count(n)
轮询索引状态	read-cypher	轮询，直到所有索引的 state = 'ONLINE'

写入审核步骤——通过 MCP 执行任何批量写入前，展示：

1. 查询语句和受影响的标签
2. 根据数据源估算的行数
3. EXPLAIN 计划

等待用户确认。未经确认，切勿自动执行 CALL IN TRANSACTIONS 或 CREATE CONSTRAINT。

如果使用的不是默认数据库，务必传入 database 参数：{"code": "...", "database": "neo4j"}。

---

常见错误

错误	原因	修复方法
Couldn't load the external resource	file:/// 路径不在 Neo4j 导入目录中	将文件移至 $NEO4J_HOME/import/；检查 dbms.security.allow_csv_import_from_file_urls=true
Cannot merge node using null property value	MERGE 键解析为 null	在 MERGE 前验证 row.id IS NOT NULL；添加 WHERE row.id IS NOT NULL
toInteger() called on null	将 Null 列传入了不支持 Null 的函数	替换为 toIntegerOrNull() → toIntegerOrNull()、toFloat() → toFloatOrNull()
Node N already exists / 约束冲突发生在导入过程中	源 ID 重复	对源 CSV 去重；使用 MERGE 而非 CREATE；为约束添加 IF NOT EXISTS
堆溢出 / OutOfMemoryError	批次过大或文件过大	减小批次；改用 CALL IN TRANSACTIONS；离线导入使用 neo4j-admin
Invalid input 'IN': expected...'	使用了 PERIODIC COMMIT	将 USING PERIODIC COMMIT 替换为 CALL IN TRANSACTIONS — PERIODIC COMMIT 已在 Cypher 25 中移除
neo4j-admin：Bad input data	标头格式错误或类型不匹配	检查是否包含 :ID、:START_ID、:END_ID；检查类型化列是否能正确解析
neo4j-admin：导入无提示地失败	2025.12 之前 --bad-tolerance 的默认值为无限制	设置 --bad-tolerance=0 以显示所有错误
MERGE 期间未使用索引	导入前未创建约束	删除数据，创建约束，然后重新导入
关系导入时缺少节点	在节点导入前导入了关系	始终先导入所有节点文件，再导入任何关系文件

---

导入后验证

导入完成后 — 运行以下全部检查：

// 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;

在所有索引都达到 ONLINE 状态之前，切勿运行生产查询。

---

参考资料

• LOAD CSV — Cypher 手册 25
• CALL IN TRANSACTIONS — Cypher 手册
• neo4j-admin 数据库导入
• APOC 周期性执行
• APOC 加载过程
• GraphAcademy：导入 CSV 数据
• 索引和约束 — 类型、MERGE 锁语义、导入前检查
• Data Importer 图形界面 — 适用场景、Aura 访问、多轮导入、注意事项
• 导入后重构 — 拆分列表、提取节点、添加标签、外键验证

---

检查清单

• 在任何基于 MERGE 的导入之前创建唯一性约束
• 如果使用 apoc.* 过程，已验证 APOC 可用
• 已确认目标为 PRIMARY（而非副本）
• 已验证源文件：UTF-8 编码、行数符合预期、无 BOM
• LOAD CSV 使用 toIntegerOrNull() / toFloatOrNull() — 绝不直接使用 toInteger()/toFloat()
• 在空字符串 ≠ null 的情况下应用 nullIf(row.x, '')
• 使用 CALL IN TRANSACTIONS（而非 USING PERIODIC COMMIT）
• 生产环境加载使用 ON ERROR CONTINUE + REPORT STATUS
• 节点导入已在关系导入之前完成
• neo4j-admin：已设置 --bad-tolerance=0；SSD 使用 --high-parallel-io=on
• 导入后：行数与源数据匹配；所有索引均为 ONLINE
• 已应用写入执行门控（MCP）：展示了查询和估算，并获得确认
• 凭据存放在 .env 中；.env 已列入 .gitignore