📚 课程目录第3章图谱构建实战

3. 图谱构建实战：使用Python操作Neo4j、批量导入节点与关系、索引与约束设置

本章进入反欺诈图谱构建的核心实操环节。我们将使用Python驱动Neo4j，完成从环境准备、批量数据导入到索引与约束设置的完整流程。掌握这些技能后，你就能将清洗后的交易、设备、账户等实体数据，高效地转化为可供图算法与XGBoost模型直接调用的图谱结构。

3.1 环境准备与Python驱动安装

在开始构建图谱之前，需要确保Neo4j数据库已启动，并安装Python的官方驱动。推荐使用Neo4j 4.x或5.x版本，驱动使用neo4j包。

  环境要求清单
  Neo4j Server：已安装并运行，默认端口7687（Bolt协议）。
Python版本：3.8及以上。
Python驱动：neo4j 5.x。
辅助库：pandas（数据预处理）、py2neo（可选，用于快速原型）。

# 安装neo4j驱动
pip install neo4j pandas

# 验证连接
from neo4j import GraphDatabase

uri = "bolt://localhost:7687"
driver = GraphDatabase.driver(uri, auth=("neo4j", "your_password"))

def test_connection(driver):
    with driver.session() as session:
        result = session.run("RETURN 'Neo4j Connected' AS message")
        for record in result:
            print(record["message"])

test_connection(driver)
# 输出: Neo4j Connected

3.2 节点与关系的批量导入策略

反欺诈图谱通常涉及百万级节点与千万级关系。逐条插入效率极低，必须采用批量导入策略。以下对比三种主流方式：

导入方式	适用场景	速度	事务控制	推荐度
UNWIND + CREATE	中等规模（10万级节点）	中等	单次事务提交	⭐⭐⭐⭐
LOAD CSV	大规模（百万级），文件已就绪	快	自动分批	⭐⭐⭐⭐⭐
neo4j-admin import	超大规模（千万级），首次全量	极快	离线导入	⭐⭐⭐

在反欺诈实时回放场景中，我们优先使用UNWIND + CREATE方式，因为它能灵活地与Python内存中的DataFrame对接，适合增量更新。

3.2.1 批量创建节点（UNWIND模式）

def batch_create_nodes(tx, node_type, properties_list, batch_size=1000):
    """
    批量创建指定类型的节点
    :param tx: 事务对象
    :param node_type: 节点标签，如 "Account"
    :param properties_list: 属性字典列表
    :param batch_size: 每批大小
    """
    for i in range(0, len(properties_list), batch_size):
        batch = properties_list[i:i+batch_size]
        query = f"""
        UNWIND $batch AS row
        CREATE (n:{node_type})
        SET n = row
        """
        tx.run(query, batch=batch)

# 示例：批量创建账户节点
accounts = [
    {"account_id": "A001", "name": "张三", "risk_score": 0.3},
    {"account_id": "A002", "name": "李四", "risk_score": 0.8},
    {"account_id": "A003", "name": "王五", "risk_score": 0.1},
]

with driver.session() as session:
    session.execute_write(batch_create_nodes, "Account", accounts)

3.2.2 批量创建关系（UNWIND模式）

def batch_create_relationships(tx, rel_type, start_label, start_key, end_label, end_key, rels_list, batch_size=1000):
    """
    批量创建关系
    :param rel_type: 关系类型，如 "TRANSFER_TO"
    :param start_label: 起始节点标签
    :param start_key: 起始节点匹配属性名
    :param end_label: 目标节点标签
    :param end_key: 目标节点匹配属性名
    :param rels_list: 关系字典列表，需包含起始/目标键值及关系属性
    """
    for i in range(0, len(rels_list), batch_size):
        batch = rels_list[i:i+batch_size]
        query = f"""
        UNWIND $batch AS row
        MATCH (a:{start_label} {{{start_key}: row.start_id}})
        MATCH (b:{end_label} {{{end_key}: row.end_id}})
        CREATE (a)-[r:{rel_type} SET r += row.properties]->(b)
        """
        tx.run(query, batch=batch)

# 示例：批量创建转账关系
transfers = [
    {"start_id": "A001", "end_id": "A002", "properties": {"amount": 5000, "timestamp": "2024-01-15"}},
    {"start_id": "A002", "end_id": "A003", "properties": {"amount": 2000, "timestamp": "2024-01-16"}},
]

with driver.session() as session:
    session.execute_write(batch_create_relationships, "TRANSFER_TO", "Account", "account_id", "Account", "account_id", transfers)

性能优化提示：批量大小建议设置在500~2000之间。过小则事务开销大，过大会导致内存溢出或事务超时。对于关系创建，务必确保MATCH的节点属性已建立索引，否则会触发全表扫描，严重拖慢导入速度。

3.3 索引与约束设置

索引是图数据库查询性能的基石，约束则保证数据完整性。在反欺诈图谱中，我们通常对实体唯一标识符建立唯一约束，对频繁查询的属性建立索引。

3.3.1 唯一约束（Unique Constraint）

唯一约束确保某个标签下指定属性的值全局唯一，同时自动为该属性创建索引。这是防止重复节点插入的关键手段。

def create_unique_constraints(driver):
    with driver.session() as session:
        # 账户ID唯一约束
        session.run("CREATE CONSTRAINT account_id_unique IF NOT EXISTS FOR (a:Account) REQUIRE a.account_id IS UNIQUE")
        # 设备指纹唯一约束
        session.run("CREATE CONSTRAINT device_fp_unique IF NOT EXISTS FOR (d:Device) REQUIRE d.fingerprint IS UNIQUE")
        # IP地址唯一约束
        session.run("CREATE CONSTRAINT ip_unique IF NOT EXISTS FOR (i:IP) REQUIRE i.address IS UNIQUE")
        print("唯一约束创建完成")

create_unique_constraints(driver)

3.3.2 单属性索引（Single-Property Index）

对于非唯一但频繁用于MATCH、WHERE或ORDER BY的属性，建立普通索引。

def create_indexes(driver):
    with driver.session() as session:
        # 账户风险评分索引（用于范围查询）
        session.run("CREATE INDEX account_risk_idx IF NOT EXISTS FOR (a:Account) ON (a.risk_score)")
        # 交易时间戳索引（用于时间窗口查询）
        session.run("CREATE INDEX transfer_time_idx IF NOT EXISTS FOR ()-[r:TRANSFER_TO]-() ON (r.timestamp)")
        # 设备最后活跃时间索引
        session.run("CREATE INDEX device_last_seen_idx IF NOT EXISTS FOR (d:Device) ON (d.last_seen)")
        print("索引创建完成")

create_indexes(driver)

3.3.3 复合索引（Composite Index）

当查询条件同时涉及多个属性时，复合索引能显著提升性能。例如，反欺诈中常按“设备类型+风险等级”组合查询。

# 复合索引示例：设备类型 + 风险等级
session.run("""
CREATE INDEX device_type_risk_idx IF NOT EXISTS 
FOR (d:Device) ON (d.device_type, d.risk_level)
""")

索引使用避坑：

不要为每个属性都建索引，索引会占用内存并拖慢写入速度。
复合索引的列顺序很重要：将选择性高（唯一值多）的属性放在前面。
对于关系属性索引，语法与节点索引不同，需使用FOR ()-[r:REL_TYPE]-() ON (r.prop)。

3.4 图谱构建完整流程SVG

下图展示了从原始数据到图谱就绪的完整构建流水线，包含数据预处理、批量导入、索引约束设置三个核心阶段。

3.5 完整实战代码模板

以下是一个整合了上述所有步骤的实战模板，可直接用于反欺诈图谱的初始化构建。

from neo4j import GraphDatabase
import pandas as pd

class FraudGraphBuilder:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def setup_schema(self):
        """设置索引与约束"""
        with self.driver.session() as session:
            # 唯一约束
            session.run("CREATE CONSTRAINT IF NOT EXISTS FOR (a:Account) REQUIRE a.account_id IS UNIQUE")
            session.run("CREATE CONSTRAINT IF NOT EXISTS FOR (d:Device) REQUIRE d.fingerprint IS UNIQUE")
            session.run("CREATE CONSTRAINT IF NOT EXISTS FOR (i:IP) REQUIRE i.address IS UNIQUE")
            # 索引
            session.run("CREATE INDEX IF NOT EXISTS FOR (a:Account) ON (a.risk_score)")
            session.run("CREATE INDEX IF NOT EXISTS FOR ()-[r:TRANSFER_TO]-() ON (r.timestamp)")
            print("Schema setup complete.")
    
    def import_nodes(self, df, node_type, batch_size=1000):
        """从DataFrame批量导入节点"""
        records = df.to_dict('records')
        with self.driver.session() as session:
            for i in range(0, len(records), batch_size):
                batch = records[i:i+batch_size]
                session.run(f"""
                    UNWIND $batch AS row
                    CREATE (n:{node_type})
                    SET n = row
                """, batch=batch)
            print(f"Imported {len(records)} {node_type} nodes.")
    
    def import_relationships(self, df, rel_type, start_label, start_key, end_label, end_key, batch_size=1000):
        """从DataFrame批量导入关系"""
        records = df.to_dict('records')
        with self.driver.session() as session:
            for i in range(0, len(records), batch_size):
                batch = records[i:i+batch_size]
                session.run(f"""
                    UNWIND $batch AS row
                    MATCH (a:{start_label} {{{start_key}: row.start_id}})
                    MATCH (b:{end_label} {{{end_key}: row.end_id}})
                    CREATE (a)-[r:{rel_type} SET r += row.properties]->(b)
                """, batch=batch)
            print(f"Imported {len(records)} {rel_type} relationships.")
    
    def close(self):
        self.driver.close()

# 使用示例
if __name__ == "__main__":
    builder = FraudGraphBuilder("bolt://localhost:7687", "neo4j", "password")
    
    # 1. 设置索引与约束
    builder.setup_schema()
    
    # 2. 准备数据（示例DataFrame）
    accounts_df = pd.DataFrame([
        {"account_id": "A001", "name": "张三", "risk_score": 0.3},
        {"account_id": "A002", "name": "李四", "risk_score": 0.8},
    ])
    transfers_df = pd.DataFrame([
        {"start_id": "A001", "end_id": "A002", "amount": 5000, "timestamp": "2024-01-15"},
    ])
    
    # 3. 批量导入
    builder.import_nodes(accounts_df, "Account")
    builder.import_relationships(transfers_df, "TRANSFER_TO", "Account", "account_id", "Account", "account_id")
    
    builder.close()

3.6 常见问题与排查

问题现象	可能原因	解决方案
导入速度极慢	未建立索引，MATCH触发全表扫描	为匹配属性建立索引或唯一约束
事务超时	batch_size过大，事务执行时间超过配置	减小batch_size至500~1000
重复节点插入	未设置唯一约束，MERGE误用为CREATE	先建唯一约束，使用MERGE替代CREATE
关系创建失败	起始或目标节点不存在	检查数据完整性，先导入节点再导入关系
内存溢出	单次事务加载数据过多	增加分页逻辑，控制每批数据量

生产环境建议：

使用MERGE代替CREATE来避免重复节点，但需配合唯一约束使用以保证性能。
对于超大规模导入（千万级），优先使用neo4j-admin import工具进行离线全量导入，再通过Python驱动进行增量更新。
定期执行PROFILE命令分析查询计划，确保索引被正确使用。

完成本章实战后，你的Neo4j中应已构建起包含账户、设备、IP等实体及转账、登录、设备关联等关系的反欺诈图谱。下一阶段，我们将基于此图谱进行回放验证，提取时序与结构特征，为XGBoost模型提供高质量输入。