Neo4j 5.x Python 连接实战：Py2neo 实现 5 类复杂查询与批量导入
Neo4j 5.x Python 连接实战：Py2neo 实现 5 类复杂查询与批量导入
在当今数据驱动的时代，图数据库因其出色的关系处理能力而备受关注。作为图数据库领域的佼佼者，Neo4j 5.x版本带来了更强大的性能和更丰富的功能。对于Python开发者而言，Py2neo库提供了与Neo4j交互的优雅方式，让复杂图查询和批量操作变得简单高效。
1. 环境准备与基础连接
在开始之前，确保已安装Neo4j 5.x数据库和Python 3.7+环境。Py2neo的最新版本（5.x系列）针对Neo4j 5.x进行了优化，提供了更好的兼容性和性能表现。
首先安装必要的依赖：
pip install py2neo pandas numpy基础连接配置是使用Py2neo的第一步。不同于简单的连接字符串配置，生产环境中我们还需要考虑连接池、超时设置和SSL加密等细节：
from py2neo import Graph # 基础连接配置 graph = Graph( "bolt://localhost:7687", auth=("neo4j", "your_password"), secure=False, # 生产环境应设为True并配置SSL证书 max_connection_lifetime=3600, # 连接最大生命周期(秒) connection_timeout=30 # 连接超时时间(秒) ) # 验证连接是否成功 try: graph.run("RETURN 1 AS test").data() print("Neo4j连接成功") except Exception as e: print(f"连接失败: {str(e)}")对于需要高性能的场景，可以调整连接池大小：
from py2neo import Graph # 高性能连接配置 high_perf_graph = Graph( "bolt://localhost:7687", auth=("neo4j", "your_password"), max_connection_pool_size=50, # 最大连接池大小 init_connection_pool_size=10 # 初始连接池大小 )2. 复杂查询实战
2.1 多跳路径查询
路径查询是图数据库的核心能力之一。Py2neo提供了直观的方式来执行复杂的多跳查询：
from py2neo import Graph graph = Graph("bolt://localhost:7687", auth=("neo4j", "your_password")) # 查找两个人之间的所有路径（最多3跳） query = """ MATCH path = (a:Person {name: $name1})-[*1..3]-(b:Person {name: $name2}) RETURN path, length(path) AS path_length ORDER BY path_length """ result = graph.run(query, name1="Alice", name2="Bob").data() # 解析结果 for record in result: path = record["path"] print(f"路径长度: {record['path_length']}") for node in path.nodes: print(f"节点: {node['name']} ({list(node.labels)[0]})") for rel in path.relationships: print(f"关系: {rel.start_node['name']}-[{rel.type}]->{rel.end_node['name']}")2.2 聚合查询与统计
Neo4j强大的聚合功能可以轻松实现复杂的数据统计：
# 统计每个城市出生的人数及其平均年龄 query = """ MATCH (p:Person)-[:BORN_IN]->(l:Location) RETURN l.city AS city, count(p) AS population, avg(p.age) AS avg_age, percentileCont(p.age, 0.5) AS median_age ORDER BY population DESC """ result = graph.run(query).to_data_frame() # 使用Pandas进行进一步分析 print(result.describe())2.3 条件过滤与复杂WHERE子句
Py2neo支持传递参数化查询，可以构建复杂的过滤条件：
# 复杂条件查询：查找特定年龄段且有特定关系的人 query = """ MATCH (p:Person)-[r]->(other) WHERE p.age >= $min_age AND p.age <= $max_age AND type(r) IN $relationship_types AND (other:Location OR (other:Person AND other.age > p.age)) RETURN p.name AS person, type(r) AS relationship, other.name AS related_to ORDER BY p.age DESC """ params = { "min_age": 20, "max_age": 40, "relationship_types": ["FRIENDS", "MARRIED", "COLLEAGUE"] } result = graph.run(query, params).to_table()2.4 最短路径与图算法
Neo4j内置的图算法可以直接通过Py2neo调用：
# 使用Dijkstra算法查找最短路径（考虑关系权重） query = """ MATCH (start:Person {name: $start_name}), (end:Person {name: $end_name}) CALL gds.shortestPath.dijkstra.stream({ nodeQuery: 'MATCH (p:Person) RETURN id(p) AS id', relationshipQuery: 'MATCH (p1:Person)-[r]->(p2:Person) RETURN id(p1) AS source, id(p2) AS target, r.weight AS weight', startNode: start, endNode: end, relationshipWeightProperty: 'weight' }) YIELD index, sourceNode, targetNode, totalCost, nodeIds, costs, path RETURN totalCost, [nodeId IN nodeIds | gds.util.asNode(nodeId).name] AS nodeNames """ result = graph.run(query, start_name="Alice", end_name="Bob").data()2.5 时间序列查询
对于带有时间属性的数据，可以执行复杂的时间序列分析：
# 查询关系随时间的变化 query = """ MATCH (p1:Person)-[r]->(p2:Person) WHERE r.since >= $start_year AND r.since <= $end_year WITH p1, p2, r, r.since AS year ORDER BY year RETURN p1.name AS person1, p2.name AS person2, type(r) AS relationship, collect(year) AS years_active """ result = graph.run(query, start_year=2000, end_year=2020).to_data_frame()3. 批量数据导入优化
对于大规模数据导入，Py2neo提供了多种优化策略。以下是万级节点批量导入的最佳实践：
3.1 使用Subgraph批量导入
from py2neo import Graph, Node, Relationship, Subgraph graph = Graph("bolt://localhost:7687", auth=("neo4j", "your_password")) # 准备批量数据 people = [ Node("Person", name=f"Person_{i}", age=20+i%30) for i in range(1, 10001) ] locations = [ Node("Location", city=f"City_{i%100}", state=f"State_{i%10}") for i in range(1, 1001) ] # 创建关系 relationships = [] for i, person in enumerate(people[:1000]): relationships.append(Relationship(person, "LIVES_IN", locations[i%100])) # 使用事务批量提交 tx = graph.begin() subgraph = Subgraph(people + locations, relationships) tx.create(subgraph) graph.commit(tx)3.2 性能优化技巧
- 批量大小控制：每批1000-5000个节点为宜
- 并行导入：使用多线程/进程加速
- 索引优化：预先创建索引和约束
# 创建索引（应在导入数据前执行） graph.run("CREATE INDEX person_name_index IF NOT EXISTS FOR (p:Person) ON (p.name)") graph.run("CREATE CONSTRAINT location_unique IF NOT EXISTS FOR (l:Location) REQUIRE (l.city, l.state) IS UNIQUE") # 并行导入示例 from concurrent.futures import ThreadPoolExecutor import numpy as np def batch_import(nodes, batch_size=1000): for i in range(0, len(nodes), batch_size): batch = nodes[i:i+batch_size] tx = graph.begin() tx.create(Subgraph(batch)) graph.commit(tx) # 将数据分成4部分并行导入 with ThreadPoolExecutor(max_workers=4) as executor: node_chunks = np.array_split(people, 4) executor.map(batch_import, node_chunks)3.3 从CSV批量导入
对于已有结构化数据，可以直接从CSV导入：
import pandas as pd from py2neo import Graph, Node, Subgraph graph = Graph("bolt://localhost:7687", auth=("neo4j", "your_password")) # 读取CSV文件 df = pd.read_csv("people.csv") # 准备节点 nodes = [] for _, row in df.iterrows(): nodes.append(Node("Person", name=row['name'], age=row['age'], gender=row['gender'])) # 批量导入 batch_size = 2000 for i in range(0, len(nodes), batch_size): tx = graph.begin() tx.create(Subgraph(nodes[i:i+batch_size])) graph.commit(tx)4. 高级模式与最佳实践
4.1 事务管理
正确的使用事务可以保证数据一致性并提高性能：
from py2neo import Graph, Node graph = Graph("bolt://localhost:7687", auth=("neo4j", "your_password")) # 事务最佳实践 try: tx = graph.begin() # 创建节点 alice = Node("Person", name="Alice", age=30) tx.create(alice) # 执行查询 tx.run("MATCH (p:Person {name: 'Bob'}) SET p.age = 31") # 提交事务 graph.commit(tx) except Exception as e: print(f"操作失败: {str(e)}") graph.rollback(tx)4.2 数据模型优化
合理的数据模型设计对性能影响巨大：
| 设计考虑 | 推荐做法 | 不推荐做法 |
|---|---|---|
| 节点标签 | 使用有意义的标签分类 | 使用单一标签或过多标签 |
| 关系类型 | 使用动词短语明确关系含义 | 使用模糊的关系类型 |
| 属性设计 | 将频繁查询的字段设为属性 | 将大文本数据存储为属性 |
| 索引策略 | 为高频查询条件创建索引 | 为所有属性创建索引 |
4.3 查询性能优化
- PROFILE查询分析：
result = graph.run("PROFILE MATCH (p:Person)-[:FRIENDS_WITH]->(f) RETURN p.name, count(f)").data()- 使用参数化查询：
# 好：参数化查询 graph.run("MATCH (p:Person) WHERE p.name = $name RETURN p", name="Alice") # 不好：字符串拼接 graph.run(f"MATCH (p:Person) WHERE p.name = 'Alice' RETURN p")- 限制结果集大小：
# 限制返回结果数量 graph.run("MATCH (p:Person) RETURN p LIMIT 100")5. 实战案例：社交网络分析
让我们通过一个完整的社交网络分析案例来综合运用上述技术：
from py2neo import Graph, Node, Relationship, Subgraph import pandas as pd import numpy as np # 初始化连接 graph = Graph("bolt://localhost:7687", auth=("neo4j", "your_password")) # 清空现有数据 graph.delete_all() # 1. 创建测试数据 num_users = 500 num_cities = 50 # 创建用户节点 users = [ Node("User", id=i, name=f"User_{i}", age=np.random.randint(18, 70), gender=np.random.choice(["M", "F"])) for i in range(num_users) ] # 创建城市节点 cities = [ Node("City", id=i, name=f"City_{i}", population=np.random.randint(10000, 1000000)) for i in range(num_cities) ] # 创建用户关系 relationships = [] for i in range(num_users): # 每个用户有3-10个朋友 friends = np.random.choice( [x for x in range(num_users) if x != i], size=np.random.randint(3, 10), replace=False ) for friend in friends: relationships.append( Relationship(users[i], "FRIENDS_WITH", users[friend]) ) # 每个用户住在1个城市 relationships.append( Relationship(users[i], "LIVES_IN", cities[np.random.randint(0, num_cities)]) ) # 批量导入数据 tx = graph.begin() tx.create(Subgraph(users + cities, relationships)) graph.commit(tx) # 2. 复杂查询：查找潜在推荐好友（朋友的朋友但不是自己的朋友） query = """ MATCH (u:User {id: $user_id})-[:FRIENDS_WITH]->(f)-[:FRIENDS_WITH]->(fof) WHERE NOT (u)-[:FRIENDS_WITH]->(fof) AND u <> fof RETURN fof.id AS user_id, fof.name AS user_name, count(f) AS mutual_friends ORDER BY mutual_friends DESC LIMIT 10 """ # 为每个用户生成好友推荐 recommendations = {} for user in users[:10]: # 只为前10个用户生成推荐 result = graph.run(query, user_id=user["id"]).data() recommendations[user["id"]] = result # 3. 社交网络分析：计算每个城市的用户年龄分布 age_distribution = graph.run(""" MATCH (u:User)-[:LIVES_IN]->(c:City) RETURN c.name AS city, count(u) AS user_count, avg(u.age) AS avg_age, percentileCont(u.age, 0.25) AS age_25, percentileCont(u.age, 0.5) AS median_age, percentileCont(u.age, 0.75) AS age_75 ORDER BY user_count DESC """).to_data_frame() # 4. 查找最活跃用户（拥有最多朋友） most_active = graph.run(""" MATCH (u:User)-[:FRIENDS_WITH]->(f) RETURN u.id, u.name, count(f) AS friend_count ORDER BY friend_count DESC LIMIT 10 """).to_table()在实际项目中，Py2neo的这些高级用法可以显著提高开发效率。记得根据具体业务需求调整查询和数据模型，并定期使用PROFILE分析查询性能。