把 CSV/JSON/Parquet 结构化数据批量导入…

在需要把结构化文件或关系型数据批量导入 Neo4j 时，按数据量/在线状态选导入方式，先建约束再分批写入，最后校验完整性。

适用场景

解决思路

1. 按决策表选方法：任意规模在线 + CSV（Aura 或本地）用 LOAD CSV + CALL IN TRANSACTIONS；< 1M 行在线 + 列表/API 响应用 UNWIND + CALL IN TRANSACTIONS；> 10M 行且离线（本地/自管理）用 `neo4j-admin database import full`；任意规模在线且有 APOC 用 `apoc.periodic.iterate` + `apoc.load.csv`；任意规模在线 + JSON/API 用 `apoc.load.json` 或 driver 批处理；增量 delta 且离线（Enterprise）用 `neo4j-admin database import incremental`。注意 Aura 只支持 `https://` URL，不支持 `file:///`，neo4j-admin import 仅用于自管理实例。
2. 导入前按顺序执行检查：先建唯一性约束（如 `CREATE CONSTRAINT IF NOT EXISTS FOR (n:Person) REQUIRE n.id IS UNIQUE;`），约束会创建 MERGE 使用的隐式 RANGE 索引并强制唯一；非唯一索引（TEXT、非键属性 RANGE、FULLTEXT）在导入后再建，Neo4j 会异步填充，轮询 `populationPercent` 到 100%；导入前建额外索引只会拖慢写入。
3. 若使用 apoc.* 过程，先执行 `RETURN apoc.version();` 验证 APOC 是否安装，失败则改用普通 LOAD CSV。
4. 确认目标库是 PRIMARY 而非副本：执行 `CALL dbms.cluster.role() YIELD role RETURN role;`，若 role ≠ `PRIMARY` 则停止并把写入重定向到 PRIMARY 端点。
5. 导入前统计源文件行数以及早发现编码问题：`wc -l data/persons.csv`。
6. 验证 UTF-8 编码（LOAD CSV 要求 UTF-8）：用 `file -i persons.csv` 检查，必要时用 `iconv -f latin1 -t utf-8 persons.csv > persons_utf8.csv` 重新编码。
7. 节点导入用 LOAD CSV + CALL IN TRANSACTIONS，配合类型转换与空值处理：`MERGE (p:Person {id: row.id})`，`ON CREATE SET` 中用 `toIntegerOrNull(row.age)`、`toFloatOrNull(row.score)`、`toBoolean(row.active)`，日期用 `CASE WHEN row.born IS NOT NULL AND row.born <> '' THEN date(row.born) ELSE null END`，`ON MATCH SET p.updatedAt = datetime()`，并以 `IN TRANSACTIONS OF 10000 ROWS ON ERROR CONTINUE REPORT STATUS AS s` 结束。
8. 处理空值规则：CSV 缺失列 → `null`（安全）；CSV 空字符串 `""` 会存为 `""` 而非 `null`，用 `nullIf(row.x, '')` 转换；`toInteger(null)`/`toFloat(null)` 会抛错，必须用 `toIntegerOrNull()`/`toFloatOrNull()`；Neo4j 不存储 `null` 属性，SET 时会被静默丢弃。
9. 关系导入（节点必须先全部存在）：`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` 结束；务必先导入所有节点再导入任何关系，否则 MATCH 会因节点缺失失败。
10. 自定义分隔符（如 TSV）：`LOAD CSV WITH HEADERS FROM 'file:///data.tsv' AS row FIELDTERMINATOR '\t'`。
11. 压缩文件（ZIP/gzip，仅本地文件）：`LOAD CSV WITH HEADERS FROM 'file:///archive.csv.gz' AS row ...`。
12. 云存储（Enterprise Edition）：AWS S3 用 `s3://my-bucket/data/persons.csv`，GCS 用 `gs://my-bucket/persons.csv`，Azure Blob 用 `azb://account/container/persons.csv`。
13. LOAD CSV 内可用内置函数：`linenumber()` 取当前行号作为兜底 ID，`file()` 取正在加载文件的绝对路径。
14. CALL IN TRANSACTIONS 语法：`CALL (row) { ... } 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]`。
15. 按场景选 ON ERROR 模式：`FAIL`（默认，首个错误回滚整个外层事务，用于严格全有全无导入）；`CONTINUE`（跳过失败批次继续，用于弹性批量导入，配合 REPORT STATUS 跟踪错误）；`BREAK`（首个失败批次后停止，保留已完成工作）；`RETRY`（指数退避重试，默认 30s，用于有死锁风险的并发写入）。注意 CONTINUE/BREAK 时外层事务仍成功，FAIL 不能与 REPORT STATUS 组合。
16. 并发批次用 `IN 4 CONCURRENT TRANSACTIONS OF 5000 ROWS ON ERROR RETRY FOR 30 SECONDS THEN CONTINUE REPORT STATUS AS s`；CONCURRENT 适用于键空间不重叠的读密集 MERGE，重叠写入有死锁风险，需配合 ON ERROR RETRY。
17. 用 REPORT STATUS 列跟踪：`s.started`（批次事务已开始）、`s.committed`（批次已成功提交）、`s.transactionId`（事务 ID）、`s.errorMessage`（失败时的错误详情，否则为 null）。
18. 按行数选批次大小：< 100k 用 10000；100k–1M 用 10000–50000（监控堆，快则加大）；1M–10M 用 50000–100000（有 CPU 时启用 CONCURRENT）；> 10M 在线用 50000（考虑改用 neo4j-admin import）；关系导入用 5000（每批做两次 MATCH，故更低）。
19. 离线批量导入用 neo4j-admin（最快，SSD 上 31M 节点/78M 关系约 3 分钟，DB 必须停止或不存在）：`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+ 可加 `--dry-run` 只校验不写入。
20. 节点头文件格式：`personId:ID,name,born:int,score:float,active:boolean,:LABEL`；数据文件无表头行，如 `p001,Alice,1985,9.2,true,Person`。字段含义：`:ID` 用于关系连线（默认不存为属性）；`:ID(Group)` 用于共享 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+）用分号分隔坐标，2026.06+ 可直接从原生 Parquet 列表类型导入。
21. 关系头文件格式：`:START_ID(Person),:END_ID(Movie),role,:TYPE`；数据如 `p001,tt0133093,Neo,ACTED_IN`；`:START_ID`/`:END_ID` 必须引用与节点文件相同的 `:ID` 组。
22. neo4j-admin 关键参数：`--delimiter`（默认 `,`，单字节 UTF-8 字符、`TAB`、`\ID` 或 `U+XXXX`，2026.07+ 拒绝换行符）；`--vector-delimiter`（默认 `;`，分隔 `prop:vector` 坐标，2026.06+ 强制与 `--delimiter` 和 `--quote` 不同）；`--id-type`（`STRING | INTEGER | ACTUAL`，默认 STRING）；`--bad-tolerance`（默认 -1 无限，2025.12 起变化，生产严格导入设 0）；`--threads`（默认 CPU 数，共享主机需显式设置）；`--max-off-heap-memory`（默认 90% RAM，共享主机需调低）；`--high-parallel-io`（默认 off，SSD/NVMe 设 on）；`--format`（默认 standard，>34B 节点/关系用 block）；`--overwrite-destination`（DB 已存在时必需）；`--dry-run`（2026.02+）。
23. 用 `--schema` 传 Cypher 文件（Enterprise，block 格式），导入完成后自动执行 `CREATE CONSTRAINT`/`CREATE INDEX`，约束优先创建（顺序已强制），路径可本地或远程（`s3://`、`gs://`、`https://`）：`neo4j-admin database import full --format=block --schema=schema.cypher --nodes=Person="persons_header.csv,persons.csv" neo4j`；schema 文件示例：`CREATE CONSTRAINT person_id IF NOT EXISTS FOR (n:Person) 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+）。
24. 增量导入（仅 Enterprise，需 block 存储格式，DB 需保持在线时用）分三阶段：`neo4j-admin database import incremental --stage=prepare --nodes=Person=persons_header.csv,delta.csv --force neo4j`；`neo4j-admin database import incremental --stage=build neo4j`（此阶段 DB 可只读）；`neo4j-admin database import incremental --stage=merge neo4j`（短暂写锁）。
25. APOC 可用时用 `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，失败批次重试 N 次，间隔 100ms）。新代码优先用原生 `CALL IN TRANSACTIONS`，它内置 REPORT STATUS、CONCURRENT、RETRY 且不依赖 APOC。
26. 用 `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`。
27. 用 `apoc.load.json` 从文件或 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`。
28. 源不是文件（API 响应、DB 迁移）时用 driver 批处理：收集成 `BATCH_SIZE`（10000）的列表，每批调用 `UNWIND $rows AS row MERGE ...`，比逐行快约 10 倍。
29. 通过 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'`。任何批量写入前展示：查询 + 受影响标签、源估算行数、`EXPLAIN` 计划，等待用户确认，绝不自动执行 `CALL IN TRANSACTIONS` 或 `CREATE CONSTRAINT`；非默认库要传 `database` 参数如 `{"code": "...", "database": "neo4j"}`。
30. 按常见错误表排查：`Couldn't load the external resource` → 文件不在 Neo4j import 目录，移到 `$NEO4J_HOME/import/` 并检查 `dbms.security.allow_csv_import_from_file_urls=true`；`Cannot merge node using null property value` → MERGE 键为 null，MERGE 前加 `WHERE row.id IS NOT NULL`；`toInteger() called on null` → 换成 `toIntegerOrNull()`/`toFloatOrNull()`；约束冲突/节点已存在 → 源 CSV 去重、用 MERGE 而非 CREATE、约束加 `IF NOT EXISTS`；堆溢出 → 减小批次、改用 CALL IN TRANSACTIONS、离线用 neo4j-admin；`Invalid input 'IN'` → 用了已移除的 `USING PERIODIC COMMIT`，改为 `CALL IN TRANSACTIONS`；neo4j-admin `Bad input data` → 检查 `:ID`/`:START_ID`/`:END_ID` 是否存在及类型列能否解析；neo4j-admin 静默失败 → 设 `--bad-tolerance=0` 暴露所有错误；MERGE 未用索引 → 导入前未建约束，需删数据、建约束、重导；关系导入缺节点 → 先导入所有节点文件再导入关系文件。
31. 导入后校验全部执行：`MATCH (n:Person) RETURN count(n) AS persons;`、`MATCH ()-[:KNOWS]->() RETURN count(*) AS knows_rels;`；再建额外非唯一索引（异步填充）如 `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);`；轮询填充进度 `SHOW INDEXES YIELD name, state, populationPercent WHERE state <> 'ONLINE' OR populationPercent < 100 RETURN name, state, populationPercent ORDER BY populationPercent;`；抽查空键 `MATCH (p:Person) WHERE p.id IS NULL RETURN count(p) AS missing_id;`；所有索引 ONLINE 前不要跑生产查询。
32. 按检查清单逐项确认：MERGE 导入前已建唯一性约束；用 apoc.* 前已验证 APOC；目标确认为 PRIMARY；源文件已验证 UTF-8、行数、无 BOM；LOAD CSV 用 `toIntegerOrNull()`/`toFloatOrNull()` 而非裸 `toInteger()`/`toFloat()`；空字符串处用 `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` 中。

适用边界 · 注意事项

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