Neo4j 数据导入导出

数据导入导出是Neo4j数据库运维的重要环节，涉及数据迁移、备份恢复、数据同步等场景。Neo4j提供了多种数据导入导出工具和方法，本文将详细介绍这些方法的使用和最佳实践。

CSV批量导入

CSV（Comma-Separated Values）是Neo4j支持的主要批量导入格式之一，适用于大规模数据导入。

1. 数据准备

节点CSV文件格式

节点CSV文件需要包含一个唯一标识符和节点属性。

csv

# 示例：persons.csv
id:ID,name,age,email
1,John Doe,30,john.doe@example.com
2,Jane Smith,25,jane.smith@example.com
3,Bob Johnson,35,bob.johnson@example.com

关系CSV文件格式

关系CSV文件需要包含起始节点ID、结束节点ID、关系类型和关系属性。

csv

# 示例：relationships.csv
:START_ID,:END_ID,:TYPE,startDate
1,2,KNOWS,2020-01-01
1,3,KNOWS,2019-05-15
2,3,FRIEND_OF,2021-03-10

2. 使用LOAD CSV导入

LOAD CSV是Neo4j的Cypher命令，用于从CSV文件导入数据。

导入节点

cypher

# 导入Person节点
LOAD CSV WITH HEADERS FROM 'file:///persons.csv' AS row
CREATE (p:Person {
  id: toInteger(row.id),
  name: row.name,
  age: toInteger(row.age),
  email: row.email
});

导入关系

cypher

# 先创建索引，提高关系导入速度
CREATE INDEX FOR (p:Person) ON (p.id);

# 导入关系
LOAD CSV WITH HEADERS FROM 'file:///relationships.csv' AS row
MATCH (a:Person {id: toInteger(row.`START_ID`)}), (b:Person {id: toInteger(row.`END_ID`)})
CREATE (a)-[r:KNOWS {
  startDate: date(row.startDate)
}]->(b);

导入选项

cypher

# 使用PERIODIC COMMIT批量提交
USING PERIODIC COMMIT 1000
LOAD CSV WITH HEADERS FROM 'file:///large-file.csv' AS row
CREATE (n:Node {property: row.property});

# 使用SKIP跳过标题行
LOAD CSV FROM 'file:///no-headers.csv' AS row SKIP 1
CREATE (n:Node {property: row[0]});

3. 使用neo4j-admin import工具

neo4j-admin import是Neo4j的命令行工具，用于大规模数据导入，速度比LOAD CSV快很多。

基本使用

bash

# 停止Neo4j服务
neo4j stop

# 使用neo4j-admin import导入数据
neo4j-admin import --database=neo4j --nodes=persons.csv --relationships=relationships.csv

# 重启Neo4j服务
neo4j start

高级选项

bash

# 指定多个节点和关系文件
neo4j-admin import \
  --database=neo4j \
  --nodes=persons-header.csv,persons1.csv,persons2.csv \
  --nodes=companies-header.csv,companies.csv \
  --relationships=works-header.csv,works.csv \
  --relationships=knows-header.csv,knows.csv

# 忽略重复节点
neo4j-admin import \
  --database=neo4j \
  --nodes=persons.csv \
  --relationships=relationships.csv \
  --ignore-duplicate-nodes

# 忽略重复关系
neo4j-admin import \
  --database=neo4j \
  --nodes=persons.csv \
  --relationships=relationships.csv \
  --ignore-duplicate-relationships

APOC插件导入

APOC（Awesome Procedures On Cypher）是Neo4j的扩展库，提供了丰富的数据导入功能。

1. 安装APOC插件

bash

# 下载APOC插件
wget https://github.com/neo4j-contrib/neo4j-apoc-procedures/releases/download/5.15.0/apoc-5.15.0-all.jar

# 将APOC插件复制到plugins目录
cp apoc-5.15.0-all.jar /var/lib/neo4j/plugins/

# 修改配置文件，允许APOC扩展
cat << EOF >> /etc/neo4j/neo4j.conf
dbms.security.procedures.unrestricted=apoc.*
dbms.security.procedures.allowlist=apoc.*
EOF

# 重启Neo4j服务
neo4j restart

2. 使用APOC导入JSON

cypher

# 从URL导入JSON
CALL apoc.load.json('https://example.com/data.json') YIELD value
CREATE (n:Node) SET n = value;

# 从本地文件导入JSON
CALL apoc.load.json('file:///data.json') YIELD value
UNWIND value.persons AS person
CREATE (p:Person) SET p = person;

3. 使用APOC导入XML

cypher

# 从URL导入XML
CALL apoc.load.xml('https://example.com/data.xml') YIELD value
UNWIND value._children AS person
CREATE (p:Person {
  name: person.name._text,
  age: toInteger(person.age._text)
});

# 从本地文件导入XML
CALL apoc.load.xml('file:///data.xml') YIELD value
CREATE (n:Node) SET n = value;

4. 使用APOC导入关系型数据库

cypher

# 从MySQL导入数据
CALL apoc.load.jdbc('jdbc:mysql://localhost:3306/mydb?user=root&password=secret', 'SELECT * FROM persons') YIELD row
CREATE (p:Person) SET p = row;

# 从PostgreSQL导入数据
CALL apoc.load.jdbc('jdbc:postgresql://localhost:5432/mydb?user=postgres&password=secret', 'SELECT * FROM persons') YIELD row
CREATE (p:Person) SET p = row;

数据导出

Neo4j提供了多种数据导出方法，用于数据备份、迁移和分析。

1. 使用cypher-shell导出CSV

bash

# 导出节点数据到CSV
cypher-shell -u neo4j -p password -d neo4j "MATCH (p:Person) RETURN p.id, p.name, p.age, p.email" --format plain --header-fields id,name,age,email > persons-export.csv

# 导出关系数据到CSV
cypher-shell -u neo4j -p password -d neo4j "MATCH (a:Person)-[r:KNOWS]->(b:Person) RETURN a.id AS start_id, b.id AS end_id, type(r) AS type, r.startDate" --format plain --header-fields start_id,end_id,type,startDate > relationships-export.csv

2. 使用APOC导出数据

导出到CSV

cypher

# 导出节点到CSV
CALL apoc.export.csv.nodes(['Person'], 'persons-export.csv', {})
YIELD file, source, format, nodes, relationships, properties, time, rows, batchSize, batches, done, data
RETURN file, rows, batchSize, batches, done;

# 导出关系到CSV
CALL apoc.export.csv.relationships(['KNOWS'], 'relationships-export.csv', {})
YIELD file, source, format, nodes, relationships, properties, time, rows, batchSize, batches, done, data
RETURN file, rows, batchSize, batches, done;

# 导出整个数据库到CSV
CALL apoc.export.csv.all('database-export.csv', {})
YIELD file, source, format, nodes, relationships, properties, time, rows, batchSize, batches, done, data
RETURN file, nodes, relationships, time, done;

导出到JSON

cypher

# 导出节点到JSON
CALL apoc.export.json.nodes(['Person'], 'persons-export.json', {})
YIELD file, source, format, nodes, relationships, properties, time, rows, batchSize, batches, done, data
RETURN file, rows, batchSize, batches, done;

# 导出查询结果到JSON
CALL apoc.export.json.query('MATCH (p:Person)-[r:KNOWS]->(f:Person) RETURN p, r, f', 'query-result.json', {})
YIELD file, source, format, nodes, relationships, properties, time, rows, batchSize, batches, done, data
RETURN file, rows, batchSize, batches, done;

3. 使用neo4j-admin dump工具

neo4j-admin dump用于创建数据库的完整备份，包括数据和元数据。

bash

# 停止Neo4j服务
neo4j stop

# 备份数据库
eo4j-admin dump --database=neo4j --to=neo4j-backup.dump

# 恢复数据库
neo4j-admin load --database=neo4j --from=neo4j-backup.dump

# 重启Neo4j服务
neo4j start

4. 使用neo4j-admin database dump工具（Neo4j 5.x+）

在Neo4j 5.x及以上版本，使用neo4j-admin database dump命令进行备份。

bash

# 在线备份数据库
neo4j-admin database dump --database=neo4j --to=neo4j-backup.dump

# 恢复数据库
neo4j-admin database load --database=neo4j --from=neo4j-backup.dump

# 增量备份
neo4j-admin database dump --database=neo4j --to=neo4j-incremental-backup.dump --incremental-from=neo4j-full-backup.dump

数据迁移策略

1. 全量迁移

全量迁移适用于初始数据迁移，将源数据库的所有数据迁移到Neo4j。

步骤

分析源数据库结构，设计Neo4j数据模型
准备源数据，转换为适合Neo4j导入的格式
使用neo4j-admin import或APOC工具导入数据
验证导入结果，确保数据完整性
执行性能测试，优化查询和索引

2. 增量迁移

增量迁移适用于持续数据同步，将源数据库的增量变化同步到Neo4j。

方法

时间戳同步：基于时间戳字段，只同步最近更新的数据
变更数据捕获（CDC）：捕获源数据库的变更日志，实时同步到Neo4j
轮询同步：定期查询源数据库，检测和同步变化
事件驱动：通过消息队列（如Kafka）实现事件驱动的数据同步

示例：基于时间戳的增量同步

cypher

# 导入最近一天更新的Person数据
USING PERIODIC COMMIT 1000
LOAD CSV WITH HEADERS FROM 'file:///persons-incremental.csv' AS row
WHERE date(row.updatedAt) >= date()-duration('P1D')
MERGE (p:Person {id: toInteger(row.id)})
SET p.name = row.name, p.age = toInteger(row.age), p.email = row.email, p.updatedAt = date(row.updatedAt);

导入性能优化

1. 硬件优化

使用高性能存储：导入过程对I/O要求较高，建议使用SSD或NVMe存储
充足的内存：确保有足够的内存用于缓存导入数据
多核CPU：导入过程可以利用多核CPU并行处理

2. 数据准备优化

预处理数据：在导入前清理和转换数据，减少导入时的计算
拆分大文件：将超大CSV文件拆分为多个小文件，提高并行处理效率
排序数据：按节点ID排序关系数据，减少磁盘寻道

3. 导入参数优化

neo4j-admin import优化

bash

# 增加并行度
neo4j-admin import --database=neo4j --nodes=persons.csv --relationships=relationships.csv --parallel=4

# 调整页面大小
neo4j-admin import --database=neo4j --nodes=persons.csv --relationships=relationships.csv --page-size=32768

# 禁用验证（仅用于可信数据）
neo4j-admin import --database=neo4j --nodes=persons.csv --relationships=relationships.csv --skip-bad-relationships

LOAD CSV优化

cypher

# 使用适当的批量大小
USING PERIODIC COMMIT 5000
LOAD CSV WITH HEADERS FROM 'file:///large-file.csv' AS row
CREATE (n:Node {property: row.property});

# 预先创建索引
CREATE INDEX FOR (n:Node) ON (n.id);

# 使用MERGE时添加约束
CREATE CONSTRAINT FOR (n:Node) REQUIRE n.id IS UNIQUE;

4. 导入后优化

bash

# 重建索引
neo4j-admin database index --database=neo4j rebuild-all

# 优化存储
neo4j-admin database compact --database=neo4j

# 分析数据库统计信息
neo4j-admin database stats --database=neo4j

常见问题与解决方案

1. 导入速度慢

问题现象：数据导入速度远低于预期。

解决方案：

使用neo4j-admin import替代LOAD CSV
增加硬件资源，特别是I/O性能
优化导入参数，如并行度、批量大小
预处理数据，减少导入时的计算
关闭不必要的服务和进程

2. 内存不足

问题现象：导入过程中出现内存不足错误。

解决方案：

减少批量大小
增加JVM堆内存
拆分大文件，分批导入
使用neo4j-admin import，它使用较少的内存

3. 数据重复

问题现象：导入后出现重复节点或关系。

解决方案：

使用MERGE替代CREATE
为节点添加唯一性约束
使用--ignore-duplicate-nodes和--ignore-duplicate-relationships选项
预处理数据，删除重复记录

4. 导入失败

问题现象：导入过程中出现错误，导致导入失败。

解决方案：

检查数据格式，确保符合要求
查看日志文件，定位具体错误
使用--skip-bad-relationships和--skip-duplicate-nodes选项
修复错误数据，重新导入

导入导出最佳实践

1. 数据模型设计

合理设计节点和关系：根据业务需求设计高效的数据模型
使用适当的标签和关系类型：避免过度使用通用标签和关系类型
优化属性结构：将相关属性组织为对象，减少节点属性数量
设计合适的索引：为常用查询属性创建索引

2. 数据准备

数据清洗：清理无效、重复和不一致的数据
数据转换：将源数据转换为适合Neo4j的数据类型
数据验证：验证数据完整性和格式正确性
数据备份：在导入前备份源数据和目标数据库

3. 导入策略

测试导入：先在测试环境中进行导入测试
监控导入过程：实时监控导入进度和资源使用情况
记录导入日志：保存导入日志，便于故障排查
验证导入结果：导入后验证数据完整性和正确性

4. 性能监控

监控系统资源：监控CPU、内存、磁盘I/O使用率
监控导入进度：了解导入的速度和预计完成时间
识别瓶颈：定位导入过程中的性能瓶颈
调整参数：根据监控结果调整导入参数

常见问题（FAQ）

Q1: 如何选择合适的数据导入方法？

A1: 选择数据导入方法的依据包括：

数据规模：大规模数据（>100万条）建议使用neo4j-admin import
数据来源：关系型数据库建议使用APOC工具
导入频率：频繁导入建议使用增量同步
实时性要求：高实时性要求建议使用CDC或事件驱动

Q2: 如何优化CSV导入性能？

A2: 优化CSV导入性能的方法包括：

使用neo4j-admin import工具
增加并行度
调整批量大小
预处理和排序数据
使用高性能存储

Q3: 如何从关系型数据库迁移到Neo4j？

A3: 从关系型数据库迁移到Neo4j的步骤包括：

分析源数据库结构
设计Neo4j数据模型
准备迁移脚本
执行数据迁移
验证迁移结果
优化查询和索引

Q4: 如何导出Neo4j数据用于分析？

A4: 导出Neo4j数据用于分析的方法包括：

使用cypher-shell导出CSV
使用APOC插件导出JSON或CSV
使用neo4j-admin dump创建完整备份
使用Neo4j Bloom进行可视化分析

Q5: 如何处理导入过程中的错误？

A5: 处理导入过程中错误的方法包括：

查看日志文件，定位具体错误
修复错误数据
使用--skip-bad-relationships和--skip-duplicate-nodes选项
分批导入，定位错误数据
检查数据格式和完整性

Q6: 如何实现Neo4j与其他数据库的数据同步？

A6: 实现Neo4j与其他数据库数据同步的方法包括：

基于时间戳的增量同步
变更数据捕获（CDC）
轮询同步
事件驱动同步
使用ETL工具，如Apache NiFi、Talend等

Q7: 如何备份Neo4j数据库？

A7: 备份Neo4j数据库的方法包括：

使用neo4j-admin dump创建完整备份
使用neo4j-admin database dump（Neo4j 5.x+）
使用CSV或JSON导出数据
定期执行增量备份
使用第三方备份工具

Q8: 如何恢复Neo4j数据库？

A8: 恢复Neo4j数据库的方法包括：

使用neo4j-admin load恢复备份
使用neo4j-admin database load（Neo4j 5.x+）
使用CSV或JSON导入数据
从增量备份恢复到指定时间点
验证恢复结果，确保数据完整性

Neo4j 数据导入导出 ​

CSV批量导入 ​

1. 数据准备 ​

节点CSV文件格式 ​

关系CSV文件格式 ​

2. 使用LOAD CSV导入 ​

导入节点 ​

导入关系 ​

导入选项 ​

3. 使用neo4j-admin import工具 ​

基本使用 ​

高级选项 ​

APOC插件导入 ​

1. 安装APOC插件 ​

2. 使用APOC导入JSON ​

3. 使用APOC导入XML ​

4. 使用APOC导入关系型数据库 ​

数据导出 ​

1. 使用cypher-shell导出CSV ​

2. 使用APOC导出数据 ​

导出到CSV ​

导出到JSON ​

3. 使用neo4j-admin dump工具 ​

4. 使用neo4j-admin database dump工具（Neo4j 5.x+） ​

数据迁移策略 ​

1. 全量迁移 ​

步骤 ​

2. 增量迁移 ​

方法 ​

示例：基于时间戳的增量同步 ​

导入性能优化 ​

1. 硬件优化 ​

2. 数据准备优化 ​

3. 导入参数优化 ​

neo4j-admin import优化 ​

LOAD CSV优化 ​

4. 导入后优化 ​

常见问题与解决方案 ​

1. 导入速度慢 ​

2. 内存不足 ​

3. 数据重复 ​

4. 导入失败 ​

导入导出最佳实践 ​

1. 数据模型设计 ​

2. 数据准备 ​

3. 导入策略 ​

4. 性能监控 ​

常见问题（FAQ） ​

Q1: 如何选择合适的数据导入方法？ ​

Q2: 如何优化CSV导入性能？ ​

Q3: 如何从关系型数据库迁移到Neo4j？ ​

Q4: 如何导出Neo4j数据用于分析？ ​

Q5: 如何处理导入过程中的错误？ ​

Q6: 如何实现Neo4j与其他数据库的数据同步？ ​

Q7: 如何备份Neo4j数据库？ ​

Q8: 如何恢复Neo4j数据库？ ​

Neo4j 数据导入导出

CSV批量导入

1. 数据准备

节点CSV文件格式

关系CSV文件格式

2. 使用LOAD CSV导入

导入节点

导入关系

导入选项

3. 使用neo4j-admin import工具

基本使用

高级选项

APOC插件导入

1. 安装APOC插件

2. 使用APOC导入JSON

3. 使用APOC导入XML

4. 使用APOC导入关系型数据库

数据导出

1. 使用cypher-shell导出CSV

2. 使用APOC导出数据

导出到CSV

导出到JSON

3. 使用neo4j-admin dump工具

4. 使用neo4j-admin database dump工具（Neo4j 5.x+）

数据迁移策略

1. 全量迁移

步骤

2. 增量迁移

方法

示例：基于时间戳的增量同步

导入性能优化

1. 硬件优化

2. 数据准备优化

3. 导入参数优化

neo4j-admin import优化

LOAD CSV优化

4. 导入后优化

常见问题与解决方案

1. 导入速度慢

2. 内存不足

3. 数据重复

4. 导入失败

导入导出最佳实践

1. 数据模型设计

2. 数据准备

3. 导入策略

4. 性能监控

常见问题（FAQ）

Q1: 如何选择合适的数据导入方法？

Q2: 如何优化CSV导入性能？

Q3: 如何从关系型数据库迁移到Neo4j？

Q4: 如何导出Neo4j数据用于分析？

Q5: 如何处理导入过程中的错误？

Q6: 如何实现Neo4j与其他数据库的数据同步？

Q7: 如何备份Neo4j数据库？

Q8: 如何恢复Neo4j数据库？