基于LOAD CSV导入含列表属性CSV至Neo4j的节点与关系创建问题

现有CSV文件格式如下：

Id, OID, name, address, Parents , Children
1, mid1, ratta, hello@aa, ["mid250","mid251","mid253"], ["mid60","mid65"]
2, mid2, butta, ado@bb, ["mid350","mid365","mid320", "mid450","mid700"], ["mid20","mid25","mid30"]
3, mid3, natta, hkk@aa, ["mid50","mid311","mid543"], []

需要通过Neo4j的LOAD CSV完成以下导入操作：

• 为CSV每一行创建对应的AAA节点
• 匹配当前行OID与Parents列中的OID，创建[:PARENT]关系（如第一行需创建mid1与mid250、mid251、mid253的PARENT关系）
• 匹配当前行OID与Children列中的OID，创建[:CHILD]关系

现有Python执行Cypher的代码如下，但执行结果不符合预期：

def create_AAA(tx):
    tx.run(
        "LOAD CSV WITH HEADERS FROM 'file:///aaa.csv' AS row MERGE (e:AAA {id: row._id,OID: row.OID,address: row.address,name: row.name})"
    )

def create_parent(tx):
    tx.run(
        "LOAD CSV WITH HEADERS FROM 'file:///aaa.csv' AS row MERGE (a:AAA {OID: row.OID}) FOREACH (t in row.parents | MERGE (e:AAA {OID:t}) MERGE (a)-[:PARENT]->(e) )"
    )

def create_child(tx):
    tx.run(
        "LOAD CSV WITH HEADERS FROM 'file:///aaa.csv' AS row MERGE (a:AAA {OID: row.OID}) FOREACH (t in row.children | MERGE (e:AAA {OID:t}) MERGE (a)-[:CHILD]->(e) )"
    )

with driver.session() as session:
    session.write_transaction(create_AAA)
    session.write_transaction(create_parent)
    session.write_transaction(create_child)

1. CSV中的Parents和Children列是JSON格式的字符串，并非原生数组，无法直接用FOREACH遍历
2. 创建节点时引用了不存在的row._id字段，CSV表头是Id，应该用row.Id
3. 空数组[]需要特殊处理，避免转换失败

解决思路：优先使用apoc.convert.fromJsonList()将JSON字符串转为数组（需确保Neo4j已安装APOC插件），同时修正字段名并处理空数组场景；若无法安装APOC，可通过字符串拆分实现。

方案一：使用APOC插件（推荐） #

def create_AAA_and_relationships(tx):
    tx.run("""
        LOAD CSV WITH HEADERS FROM 'file:///aaa.csv' AS row
        // 创建当前行的AAA节点
        MERGE (a:AAA {
            id: row.Id,
            OID: row.OID,
            address: row.address,
            name: row.name
        })
        // 处理Parents列，创建PARENT关系
        WITH a, row, apoc.convert.fromJsonList(row.Parents) AS parentsList
        FOREACH (parentOID IN parentsList |
            MERGE (p:AAA {OID: parentOID})
            MERGE (a)-[:PARENT]->(p)
        )
        // 处理Children列，创建CHILD关系
        WITH a, row, apoc.convert.fromJsonList(row.Children) AS childrenList
        FOREACH (childOID IN childrenList |
            MERGE (c:AAA {OID: childOID})
            MERGE (a)-[:CHILD]->(c)
        )
    """)

with driver.session() as session:
    session.write_transaction(create_AAA_and_relationships)

方案二：无APOC插件的替代方案 #

def create_AAA_and_relationships(tx):
    tx.run("""
        LOAD CSV WITH HEADERS FROM 'file:///aaa.csv' AS row
        // 创建当前行的AAA节点
        MERGE (a:AAA {
            id: row.Id,
            OID: row.OID,
            address: row.address,
            name: row.name
        })
        // 处理Parents列：去除首尾括号和引号，拆分OID列表
        WITH a, row, 
             [x | x IN split(replace(replace(row.Parents, '[', ''), ']', ''), ', ') 
              WHERE x <> '' AND x <> '""'] AS parentsList
        FOREACH (parentOID IN parentsList |
            MERGE (p:AAA {OID: replace(parentOID, '"', '')})
            MERGE (a)-[:PARENT]->(p)
        )
        // 处理Children列：同上
        WITH a, row, 
             [x | x IN split(replace(replace(row.Children, '[', ''), ']', ''), ', ') 
              WHERE x <> '' AND x <> '""'] AS childrenList
        FOREACH (childOID IN childrenList |
            MERGE (c:AAA {OID: replace(childOID, '"', '')})
            MERGE (a)-[:CHILD]->(c)
        )
    """)

with driver.session() as session:
    session.write_transaction(create_AAA_and_relationships)

关键修正点说明 #

• 替换row._id为row.Id，匹配CSV表头字段
• 将节点创建和关系创建合并为一个事务，减少重复加载CSV的IO开销
• 对空数组做过滤处理，避免无效遍历操作
• 通过APOC或字符串处理，将JSON格式的字符串转为可遍历的OID列表

内容的提问来源于stack exchange，提问作者Nirmana