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

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

我必须从 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});

解决方法

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

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

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

编辑

因此，您在这里遇到的问题是您的 MERGE 不受唯一约束的支持。为了得到这个，约束的属性和标签必须出现在模式中，如果你只在 ON CREATE 或 ON MATCH 部分有它们，它将不起作用。 >

因此，如果您的约束在 :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。如果之前的执行时间对您来说是个问题，那可能是因为它正在执行标签扫描，这在加载时表现不佳。