Neo4j批量处理大JSON文件时CALL子查询报错求助

原本编写的Cypher脚本可读取JSON文件并创建/合并Member节点及关联关系，小文件运行正常，但处理大文件时出现内存溢出。为解决内存问题，尝试改用批量事务改造脚本，却触发语法错误。

原正常运行脚本 #

// Read the json file
CALL apoc.load.json('file:///test/test.json') YIELD value

// Read the json fields as variables
with value.globalcustid as v_globalcustid, value.member_node_properties as member_node_properties, value.relationships as relationships

// Update the member node demographic properties, if the member node does not exist, create it
MERGE (m:Member {globalcustid: v_globalcustid})
ON CREATE SET m.age_group=member_node_properties.age_group, m.gender=member_node_properties.gender, m.education=member_node_properties.education
ON MATCH SET m.age_group=member_node_properties.age_group, m.gender=member_node_properties.gender, m.education=member_node_properties.education

// Unwind the relationships array into multiple rows of relationship
with v_globalcustid, m, member_node_properties, relationships
UNWIND relationships as r

// Get the tag node and skip the null tag
with m, v_globalcustid, r, r.node.label as node_label, r.node.tag_id as tag_id
where tag_id is not null
CALL apoc.cypher.run(
    "MATCH (t:" + node_label + " {tag_id: '" + tag_id +"'}) RETURN t limit 1", 
    {}
) yield value as tag_node

// Merge the relationship of member node and tag node
with m, v_globalcustid, r, tag_node.t as tag_node
CALL apoc.merge.relationship(
  m, 
  r.relationship.label,
  {createdate: r.relationship.createdate, enddate: r.relationship.enddate, value: r.relationship.value},
  {},
  tag_node,
  {}
) YIELD rel

return *

改造后的批量事务脚本（报错版本） #

:auto

// Read the json file
CALL apoc.load.json('file:///test/test.json') YIELD value

// Read the json fields as variables
with value.globalcustid as v_globalcustid, value.member_node_properties as member_node_properties, value.relationships as relationships

call {
with v_globalcustid, member_node_properties, relationships

// Update the member node demographic properties, if the member node does not exist, create it
MERGE (m:Member {globalcustid: v_globalcustid})
ON CREATE SET m.age_group=member_node_properties.age_group, m.gender=member_node_properties.gender, m.education=member_node_properties.education
ON MATCH SET m.age_group=member_node_properties.age_group, m.gender=member_node_properties.gender, m.education=member_node_properties.education

// Unwind the relationships array into multiple rows of relationship
with v_globalcustid, m, member_node_properties, relationships
UNWIND relationships as r

// Get the tag node and skip the null tag
with m, v_globalcustid, r, r.node.label as node_label, r.node.tag_id as tag_id
where tag_id is not null
CALL apoc.cypher.run(
    "MATCH (t:" + node_label + " {tag_id: '" + tag_id +"'}) RETURN t limit 1", 
    {}
) yield value as tag_node

// Merge the relationship of member node and tag node
with m, v_globalcustid, r, tag_node.t as tag_node
CALL apoc.merge.relationship(
  m, 
  r.relationship.label,
  {createdate: r.relationship.createdate, enddate: r.relationship.enddate, value: r.relationship.value},
  {},
  tag_node,
  {}
) YIELD rel

return *
} in transactions of 10000 rows

报错信息 #

Query cannot conclude with CALL together with YIELD (line 31, column 1 (offset: 1319))
"CALL apoc.merge.relationship("
 ^

疑问：call {}块内调用其他存储过程是否不被允许？该如何解决？

报错原因 #

并非call {}子查询块内不能调用存储过程，而是子查询的最后一步不能以带YIELD的CALL语句结尾。Cypher语法要求子查询必须以RETURN、无YIELD的写操作（如CREATE/MERGE）等语句收尾，带YIELD的CALL作为子查询最终步骤会触发语法校验错误。

具体修复步骤 #

1. 给子查询添加收尾的RETURN语句：在apoc.merge.relationship的YIELD rel之后，添加return m, tag_node, rel或return *，让子查询以RETURN结尾。
2. 修复apoc.cypher.run的安全问题：原脚本通过字符串拼接节点标签和tag_id，存在注入风险，改用参数化查询传递变量，避免安全漏洞。
3. 调整批量事务行数：根据每个Member关联的relationships数量，适当调整事务行数（比如从10000改为1000），避免单事务处理数据量过大再次触发内存问题。

修复后的完整脚本 #

:auto

// Read the json file
CALL apoc.load.json('file:///test/test.json') YIELD value

// Read the json fields as variables
WITH value.globalcustid AS v_globalcustid, 
     value.member_node_properties AS member_node_properties, 
     value.relationships AS relationships

CALL {
    WITH v_globalcustid, member_node_properties, relationships

    // Update the member node demographic properties, if the member node does not exist, create it
    MERGE (m:Member {globalcustid: v_globalcustid})
    ON CREATE SET m.age_group = member_node_properties.age_group, 
                  m.gender = member_node_properties.gender, 
                  m.education = member_node_properties.education
    ON MATCH SET m.age_group = member_node_properties.age_group, 
                 m.gender = member_node_properties.gender, 
                 m.education = member_node_properties.education

    // Unwind the relationships array into multiple rows of relationship
    WITH v_globalcustid, m, relationships
    UNWIND relationships AS r

    // Get the tag node and skip the null tag
    WITH m, r, r.node.label AS node_label, r.node.tag_id AS tag_id
    WHERE tag_id IS NOT NULL
    // 使用参数化查询替代字符串拼接，避免注入风险
    CALL apoc.cypher.run(
        "MATCH (t:`$label` {tag_id: $tagId}) RETURN t LIMIT 1", 
        {label: node_label, tagId: tag_id}
    ) YIELD value AS tag_node

    // Merge the relationship of member node and tag node
    WITH m, r, tag_node.t AS tag_node
    CALL apoc.merge.relationship(
        m, 
        r.relationship.label,
        {createdate: r.relationship.createdate, enddate: r.relationship.enddate, value: r.relationship.value},
        {},
        tag_node,
        {}
    ) YIELD rel

    // 添加RETURN语句作为子查询收尾
    RETURN m, tag_node, rel
} IN TRANSACTIONS OF 1000 rows // 根据实际情况调整批量大小

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