python - 使用 Neo4j-Python 连接器创建/合并图形期间出现 CypherError

Question

您好，我在尝试创建/合并节点和关系时收到以下错误。它并不总是发生，但偶尔会发生。

CypherError: An unexpected failure occurred, see details in the database logs, reference number eaf50bff-deca-4055-9450-6a76c31534e4.

这是回溯:

---------------------------------------------------------------------------
CypherError                               Traceback (most recent call last)
<ipython-input-21-9700f3a5d3fa> in <module>()
      7     tx.success = True
      8     #tx.close()
----> 9   session.close()

/databricks/python/local/lib/python2.7/site-packages/neo4j/v1/session.pyc in close(self)
    522         """
    523         if self.last_result:
--> 524             self.last_result.buffer()
    525         if self.transaction:
    526             self.transaction.close()

/databricks/python/local/lib/python2.7/site-packages/neo4j/v1/session.pyc in buffer(self)
    246         if self.connection and not self.connection.closed:
    247             while not self._consumed:
--> 248                 self.connection.fetch()
    249             self.connection = None
    250

我的创建/合并代码是这样的:

for chunk in chunk_list: 
  with session.begin_transaction() as tx:
    for record in chunk:
      tx.run("MERGE (source:UID {userid : {m}, timestamp: {a}})"
             "MERGE (target:UID {userid : {n}, timestamp: {a}})"
             "MERGE (source)-[:HasConnection]-(target)", {"m": record.source, "n": record.target, "a": record.unix_timestamp_s})
    tx.success = True
    #tx.close()
  session.close()

chunk_list 是包含多个记录列表的列表。 chunk_list 中的每个列表都有多行(大约 10000)，每行包含三列:源、目标和时间戳。

对于 chunk_list 中的每个列表，我们打开一个 session ，然后执行合并操作，然后关闭 session 。

当图超过 1000 万个节点时，问题就会开始发生。假设第 1 天，chunk_list 有 400 万行，它会工作得很好，对于第 2 天，如果 chunk_list 有 400 万行，它也会工作得很好。但是，如果在第 3 天，Neo4j 图中有 300 万行，并且节点总数超过 1000 万，那么问题就开始发生。

Answer 1

执行时间如此长的原因之一是您的查询没有执行您认为它正在执行的操作。

执行 MERGE 就像尝试首先执行 MATCH，如果没有找到匹配项，则执行 CREATE。

在您的评论中，您说时间戳应该在 MERGE 操作期间更改，因此您真正想要做的是更新具有相同用户 ID 的节点的属性，但这不是您的 MERGE 所做的。

您的 MERGE 首先尝试将 :UID 节点与给定的用户 ID 和时间戳进行匹配，但是您的图中不会有这样的节点，因为时间戳是新的...将会有一个 :UID 节点相同的用户 ID，但具有不同的时间戳，因此不会找到匹配项，并且将使用与现有节点相同的用户 ID 和新时间戳创建一个全新的 :UID 节点。

因此，您的 MERGE 始终会创建新节点，并且永远不会匹配现有节点。 MERGE 本身永远不会更新属性值，因此永远不要尝试以这种方式使用它。

要执行匹配和更新，您需要仅根据现有节点上匹配的最小唯一属性进行 MERGE，然后使用 SET 更新属性，如下所示:

for chunk in chunk_list: 
  with session.begin_transaction() as tx:
    for record in chunk:
      tx.run("MERGE (source:UID {userid : {m}})"
             "SET source.timestamp = {a}"
             "MERGE (target:UID {userid : {n}})"
             "SET target.timestamp = {a}"
             "MERGE (source)-[:HasConnection]-(target)", {"m": record.source, "n": record.target, "a": record.unix_timestamp_s})
    tx.success = True
    #tx.close()
  session.close()

此外，为了确保 MERGE 快速，您需要对 :UID(userid) 有一个索引或唯一约束，否则 Neo4j 将必须进行标签扫描来查找节点，这太慢了图表的大小。

编辑

我还没有将 Python 与 Neo4j 一起使用，但我不确定这种循环是处理多个记录的正确方法，因为它看起来会运行给定 Cypher 段的大量副本。每行 5 行 Cypher 似乎很极端...理想情况下，您应该使用 Cypher 代码同时对所有行(或至少每个 block 的行)进行操作，并且您的输入可以轻松转换为行以供执行。

更好的方法可能是使用循环生成对象列表(具有“m”和“n”属性)，并将该列表和单个时间戳作为参数提交给查询。

然后，您可以将 Cypher 中的列表展开回行，并使用它。因此，如果“row”是您的对象列表参数，而“a”仍然是您的 unix 时间戳参数，则 Cypher 段将如下所示:

UNWIND {row} as line
MERGE (source:UID {userid : line.m})
SET source.timestamp = {a}
MERGE (target:UID {userid : line.n})
SET target.timestamp = {a}
MERGE (source)-[:HasConnection]-(target)