本文详解如何通过 unwind 批量写入、合理分批与参数化查询，显著提升 neo4j python 驱动在数十万行以上数据导入场景下的性能，避免逐行执行导致的严重延迟。
在使用 Neo4j Python 官方驱动（neo4j==5.20+）进行大规模数据写入时，常见的性能陷阱是对每一行数据单独发起一次事务（如 session.execute_write(create_entity, row)）。这种方式虽逻辑清晰，但会因频繁的网络往返、事务开销和驱动层序列化成本，导致吞吐量急剧下降——尤其当数据量超过 20 万行时，耗时可能呈线性甚至超线性增长。
根本优化思路是：减少事务次数 + 减少 Cypher 解析开销 + 利用 Neo4j 原生批量能力。Neo4j 提供的 UNWIND 子句正是为此设计：它可将一个参数列表展开为多行记录，在单次查询中完成批量创建或更新。
✅ 推荐做法：UNWIND + 分批提交（Batched UNWIND）
以下是一个生产就绪的示例，适用于 Pandas DataFrame 或任意可迭代的数据源：
Secure Python Code Sandbox
下载
安全 Python 代码沙箱：在隔离的沙箱中执行任意 Python 代码，预装 requests、NumPy、pandas 库。返回 stdout、stderr。
from neo4j import GraphDatabase
import pandas as pd
driver = GraphDatabase.driver(
"bolt://localhost:7687",
auth=(os.getenv("NEO_USERNAME"), os.getenv("NEO_PASSWORD"))
)
# 使用 UNWIND 的 MERGE 查询（支持去重 + 时间戳更新）
query = """
UNWIND $rows AS row
MERGE (e:Entity {EntityId: row.entity_id})
ON CREATE SET e.LastAccess = timestamp()
ON MATCH SET e.LastAccess = timestamp()
"""
BATCH_SIZE = 10_000 # 根据内存与网络调整，通常 5k–20k 较优
def batch_write_dataframe(df: pd.DataFrame, query: str, batch_size: int = BATCH_SIZE):
total_rows = len(df)
for start_idx in tqdm(range(0, total_rows, batch_size), desc="Writing batches"):
end_idx = min(start_idx + batch_size, total_rows)
batch_df = df.iloc[start_idx:end_idx]
# 转为字典列表（列名需与 Cypher 中 row.xxx 一致）
batch_data = batch_df[["entity_id"]].to_dict(orient="records")
try:
driver.execute_query(
query,
rows=batch_data,
database_="neo4j" # 显式指定数据库名（v5.9+ 推荐）
)
except Exception as e:
print(f"Failed batch [{start_idx}:{end_idx}]: {e}")
raise
# 调用示例
batch_write_dataframe(df, query)? 关键说明：driver.execute_query() 是 v5.0+ 推荐的顶层 API，自动管理会话与事务，比手动 session.execute_write() 更简洁且性能更优；$rows 是传入的参数名，必须与 UNWIND $rows AS row 中一致；row.entity_id 对应 DataFrame 中的列；ON CREATE/ON MATCH 确保唯一约束下安全更新，前提是已提前创建索引或约束（如 CREATE CONSTRAINT ON (e:Entity) ASSERT e.EntityId IS UNIQUE）；database_ 参数（注意末尾下划线）显式指定目标数据库，避免路由开销。
⚠️ 注意事项与调优建议
- 索引先行：在执行批量 MERGE 前，务必确保 :Entity(EntityId) 上存在唯一约束或节点索引，否则 MERGE 将退化为全表扫描，性能崩溃；
- 批量大小权衡：BATCH_SIZE 过小 → 事务过多；过大 → 单次请求内存/网络压力大、失败回滚代价高。建议从 5000 开始压测，观察 Neo4j 日志中的 QueryExecutionTime 和客户端内存占用；
- 错误处理粒度：上述示例按批失败，便于定位问题批次；若需细粒度容错（如跳过个别脏数据），可在 batch_data 中预清洗，或改用 apoc.periodic.iterate（需 APOC 插件）；
-
关系批量写入：同理，使用双 MATCH + UNWIND，例如：
UNWIND $rels AS rel MATCH (a:Entity {EntityId: rel.src_id}) MATCH (b:Entity {EntityId: rel.dst_id}) CREATE (a)-[:RELATED_TO]->(b) - 替代方案参考：对于超大数据集（千万级+），可考虑 neo4j-admin import（离线 CSV 导入）或流式 LOAD CSV（服务端执行），但需牺牲 Python 逻辑灵活性。
✅ 性能对比（典型场景）
| 方式 | 20 万行耗时 | 吞吐量 | 适用场景 |
|---|---|---|---|
| 逐行 execute_write | > 15 分钟 | ~200 行/秒 | 仅调试或极小数据 |
| UNWIND + 10k 批量 | ~35 秒 | ~5700 行/秒 | 推荐默认方案 |
| UNWIND + 50k 批量 | ~22 秒（内存稳定前提下） | ~9000 行/秒 | 生产环境高吞吐 |
通过合理采用 UNWIND 批量模式，配合参数化查询与显式数据库路由，你可轻松将大规模数据写入性能提升 10–50 倍，同时保持代码简洁性与可维护性。
立即学习“Python免费学习笔记（深入）”；
