如何使用Cypher加载CSV文件构建符合指定关联规则的Neo4j图数据库

核心思路 #

按CSV的numberRow升序逐行处理数据，过程中缓存最近出现的非空A、B节点：

• 若当前行A字段非空，新建A节点并更新缓存
• 若当前行B字段非空，新建B节点并更新缓存
• 每一行必新建C节点，关联当前缓存的最新A、B节点
• 用MERGE避免A、B之间重复创建:R关系

完整Cypher脚本（适配Neo4j 5.9及以上版本） #

// 按行号升序加载CSV，保证继承顺序正确
LOAD CSV WITH HEADERS FROM 'file:///file.csv' AS row
WITH row ORDER BY toInteger(row.numberRow) ASC
WITH collect(row) AS rows
CALL {
  WITH rows
  // 初始化缓存变量：最近的A节点、最近的B节点
  VAR lastA = null
  VAR lastB = null
  FOR row IN rows |
    // 处理当前行A节点：非空则新建，否则用缓存
    SET lastA = CASE WHEN row.A IS NOT NULL AND trim(row.A) <> '' 
                     THEN CREATE (:A {numberRow: toInteger(row.numberRow), value: row.A})
                     ELSE lastA END
    // 处理当前行B节点：非空则新建，否则用缓存
    SET lastB = CASE WHEN row.B IS NOT NULL AND trim(row.B) <> ''
                     THEN CREATE (:B {numberRow: toInteger(row.numberRow), value: row.B})
                     ELSE lastB END
    // 新建当前行C节点
    SET c = CREATE (:C {numberRow: toInteger(row.numberRow), value: row.C})
    // 创建关系：A和B之间的无向R，B到C的有向R
    MERGE (lastA)-[:R]-(lastB)
    CREATE (lastB)-[:R]->(c)
}
RETURN count(*) AS 处理完成行数

低版本Neo4j兼容方案（需安装APOC插件） #

如果使用的Neo4j版本低于5.9，不支持原生VAR和FOR循环语法，可以改用APOC的迭代+临时缓存实现：

// 先清空上一次运行可能残留的临时缓存
CALL apoc.temporary.remove("lastA")
CALL apoc.temporary.remove("lastB")

CALL apoc.periodic.iterate(
  // 第一阶段：按行号升序读取所有行
  'LOAD CSV WITH HEADERS FROM "file:///file.csv" AS row RETURN row ORDER BY toInteger(row.numberRow) ASC',
  // 第二阶段：逐行处理节点和关系
  '
  WITH row, apoc.temporary.get("lastA") AS lastA, apoc.temporary.get("lastB") AS lastB
  // 处理A节点
  WITH row,
       CASE WHEN row.A IS NOT NULL AND trim(row.A) <> "" 
            THEN CREATE (:A {numberRow: toInteger(row.numberRow), value: row.A})
            ELSE lastA END AS currentA,
       lastB
  // 处理B节点
  WITH row, currentA,
       CASE WHEN row.B IS NOT NULL AND trim(row.B) <> ""
            THEN CREATE (:B {numberRow: toInteger(row.numberRow), value: row.B})
            ELSE lastB END AS currentB
  // 更新临时缓存
  CALL apoc.temporary.set("lastA", currentA)
  CALL apoc.temporary.set("lastB", currentB)
  // 新建C节点并创建关联关系
  CREATE (c:C {numberRow: toInteger(row.numberRow), value: row.C})
  MERGE (currentA)-[:R]-(currentB)
  CREATE (currentB)-[:R]->(c)
  ',
  // 必须关闭并行、设置单批次处理1行，保证顺序执行缓存正确
  {batchSize:1, parallel:false}
)

注意事项 #

1. CSV表头大小写严格匹配，脚本中使用row.A/row.B/row.C和你提供的表头大写规则保持一致，避免读取到空值
2. 脚本中增加了trim()判断，避免CSV中存在空格填充的空单元格被误判为有效值
3. 使用APOC方案前需要确认已安装对应版本的APOC插件，且配置文件中开启了临时函数权限：dbms.security.procedures.unrestricted=apoc.*

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