使用 apoc.periodic.iterate 为 csv 加载 neo4j 创建节点时如何避免重复？

Question

我必须从 CSV 文件创建节点。由于文件太大，我正在使用 apoc.periodic.iterate 函数来读取文件。 CSV 具有重复条目，并且由于并行处理而创建了重复节点。如何确保即使使用并行处理也不会创建重复节点。

CALL apoc.periodic.iterate('LOAD CSV WITH HEADERS FROM "file:///DATA.csv" AS payload return payload',
'MERGE (l:PERSON {name :payload.name})
 ON CREATE SET 
              l.pid = payload.id,             
              l.createdDate= timestamp(),
              l.lastModifiedDate= timestamp()             
 ON MATCH SET 
              l.lastModifiedDate= timestamp()',
    {batchSize:500, parallel:true, concurrency: 4});

Answer 1

对于并行执行，您需要对 :PERSON(name) 设置唯一约束，因为唯一约束具有模式锁，可防止在这种情况下重复创建。

如果名称不是唯一的，则禁用并行导入或清理您的数据，以免重复。

您还可以在 MERGE 之前锁定单个节点以保证互斥，但这与禁用并行执行具有相同的效果，因此将并行切换为 false 是更好的选择。

编辑

所以您在这里遇到的问题是您的 MERGE 没有唯一约束支持。为了实现这一点，约束的属性和标签必须存在于模式中，如果你只在 ON CREATE 或 ON MATCH 部分中使用它们，它将不起作用。 p>

所以如果你的约束是:PERSON(pid)，那么你的模式应该是：

MERGE (l:PERSON {pid:payload.id})

然后您可以在 ON CREATE 部分设置名称。

如果名称应该是唯一的，那么您可以通过在 :PERSON(name) 上创建唯一约束来获得相同的效果

虽然您可以在查询中使用 EXPLAIN 来获取查询计划，但由于这是一个查询字符串，您需要进行一些剪切和粘贴操作以查看相关查询的计划，并进行少量添加在顶部，以便查询可以编译。你可以试试这个：

EXPLAIN
WITH $payload as payload
MERGE (l:PERSON {name :payload.name})
 ON CREATE SET 
              l.pid = payload.id,             
              l.createdDate= timestamp(),
              l.lastModifiedDate= timestamp()             
 ON MATCH SET 
              l.lastModifiedDate= timestamp()

您希望在计划中看到 NodeUniqueIndexSeek(Locking) 运算符，以便 MERGE 正常工作并使用正确的锁定来防止重复。

如果您看到NodeIndexSeek，则仅使用了一个索引，它不会防止重复，因为没有任何东西可以锁定来保证互斥。你需要那个独特的约束。

如果 NodeByLabelScan 存在，那就更糟了，因为您没有唯一约束或支持 MERGE 的索引。如果之前执行时间对您来说是个问题，那可能是因为它正在执行标签扫描，这对于加载来说根本性能不佳。