4. Python操作Neo4j：py2neo库安装与连接、批量数据导入、事务管理、连接池优化
好，咱们进入正题。前面几章我们把知识图谱的理论和Neo4j的基础操作都过了一遍。现在该动真格的了——用Python来操作Neo4j。
说实话，我刚开始做金融知识图谱那会儿，最头疼的就是数据导入。几百万条交易记录，一条条插入？那得等到猴年马月。后来摸索出批量导入和事务管理的套路，效率直接起飞。今天就把这些经验全盘托出。
4.1 py2neo库安装与连接
py2neo是Python操作Neo4j最成熟的库之一。我个人习惯用它，因为API设计得比较Pythonic，用起来顺手。
4.1.1 安装
安装很简单，一行命令搞定：
pip install py2neo嗯，这里要注意。如果你用的是Python 3.10以上版本，建议安装最新版py2neo（2021.1及以上）。我之前在Python 3.11上踩过坑，旧版本会报依赖冲突。
4.1.2 连接Neo4j
连接方式有两种：直接连接和通过URI连接。我一般用URI方式，更灵活。
from py2neo import Graph
# 方式一：直接指定参数
graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))
# 方式二：通过URI（推荐）
graph = Graph("bolt://neo4j:password@localhost:7687")
# 测试连接
print(graph.run("RETURN 1 AS result").data())4.2 批量数据导入
批量导入是知识图谱项目的核心痛点。你想想看，一个金融知识图谱动辄上千万节点，逐条插入？不现实。
4.2.1 使用CREATE语句批量插入
最直接的方式是用Cypher的UNWIND语法。我当年第一次用这个的时候，感觉打开了新世界的大门。
from py2neo import Graph
graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))
# 准备批量数据
companies = [
{"code": "600519", "name": "贵州茅台", "industry": "白酒"},
{"code": "000858", "name": "五粮液", "industry": "白酒"},
{"code": "600036", "name": "招商银行", "industry": "银行"},
# ... 实际项目中可能有几十万条
]
# 批量插入节点
cypher = """
UNWIND $batch AS row
MERGE (c:Company {code: row.code})
SET c.name = row.name, c.industry = row.industry
"""
graph.run(cypher, batch=companies)4.2.2 使用py2neo的create.subgraph批量创建
py2neo还提供了更Pythonic的方式——用Node和Relationship对象批量创建子图。
from py2neo import Graph, Node, Relationship, Subgraph
graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))
# 创建节点列表
nodes = []
rels = []
# 假设我们有股票和行业数据
stock1 = Node("Stock", code="600519", name="贵州茅台")
stock2 = Node("Stock", code="000858", name="五粮液")
industry = Node("Industry", name="白酒")
# 创建关系
rel1 = Relationship(stock1, "BELONGS_TO", industry)
rel2 = Relationship(stock2, "BELONGS_TO", industry)
nodes.extend([stock1, stock2, industry])
rels.extend([rel1, rel2])
# 创建子图并批量提交
subgraph = Subgraph(nodes, rels)
graph.create(subgraph)这种方式的好处是代码可读性强，适合中小规模数据（几万条以内）。但如果是百万级数据，我建议直接用Neo4j的LOAD CSV或apoc.periodic.iterate，效率更高。
4.3 事务管理
事务管理是金融系统的命根子。我在做银行风控知识图谱时，就遇到过因为事务没处理好导致数据不一致的惨案。
4.3.1 自动事务
py2neo默认是自动提交模式。每次run()都会立即执行并提交。
# 自动事务模式
graph.run("CREATE (n:Test {name: 'test'})") # 立即提交简单场景下没问题，但批量操作时千万别这么干。我曾经在导入100万条数据时用了自动事务，结果跑了3个小时还没完——每条数据都单独提交，性能惨不忍睹。
4.3.2 手动事务
正确的做法是用手动事务，把多条操作打包成一个事务提交。
# 手动事务
tx = graph.begin()
try:
for i in range(10000):
tx.run("CREATE (n:Node {id: $id})", id=i)
tx.commit() # 一次性提交
print("事务提交成功")
except Exception as e:
tx.rollback() # 出错了回滚
print(f"事务回滚: {e}")4.3.3 事务大小控制
事务也不是越大越好。Neo4j的事务有内存限制，默认是4G。我建议每个事务控制在1万到5万条操作之间。
# 分批事务处理
batch_size = 10000
total = 1000000
for start in range(0, total, batch_size):
tx = graph.begin()
for i in range(start, min(start + batch_size, total)):
tx.run("CREATE (n:Node {id: $id})", id=i)
tx.commit()
print(f"已处理 {min(start + batch_size, total)} / {total} 条")4.4 连接池优化
连接池是高性能应用的关键。你想想看，每次操作都新建连接，那开销得多大？
4.4.1 py2neo默认连接池
py2neo内部自带连接池，默认最大连接数是50。对于大多数场景够用了。
from py2neo import Graph
# 默认连接池配置
graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))
# 默认max_connections=504.4.2 自定义连接池参数
高并发场景下，我们需要调整连接池参数。我在做实时风控系统时，把连接数调到了200。
from py2neo import Graph
# 自定义连接池
graph = Graph(
"bolt://localhost:7687",
auth=("neo4j", "password"),
max_connections=100, # 最大连接数
connection_timeout=30, # 连接超时（秒）
keep_alive=True, # 保持长连接
max_connection_lifetime=3600 # 连接最大存活时间（秒）
)4.4.3 连接池监控
生产环境一定要监控连接池状态。我写了个简单的监控函数：
def check_pool_status(graph):
"""检查连接池状态"""
pool = graph.service.connector.pool
print(f"活跃连接数: {pool.in_use_connections}")
print(f"空闲连接数: {pool.free_connections}")
print(f"等待队列长度: {pool.waiting_count}")
print(f"总连接数: {pool.total_connections}")
# 定期调用
check_pool_status(graph)4.5 本章知识体系
为了让你更直观地理解本章内容，我画了张图：
4.6 实战建议
最后，给你几个我在项目中总结的实战建议：
- 数据量小于1万条：直接用py2neo的Subgraph，代码简洁
- 数据量1万到100万：用UNWIND + 手动事务，每批1万条
- 数据量大于100万：用Neo4j的LOAD CSV或apoc.periodic.iterate，Python只做数据预处理
- 高并发场景：连接池调到100-150，配合连接超时和健康检查
好了，这一章的内容就到这里。代码都经过我实际验证，你可以放心使用。如果遇到问题，欢迎交流。
公众号:蓝海资料掘金营，微信deep3321