database - Neo4j基于具有某些属性的其他标签创建新标签

Question

我需要创建基于点的轨迹。
轨迹可以包含满足特定条件的任意数量的点。
条件是：CamerasID、TrajectoryID、ClassType和ClassQual应该相等。
每个点的时间差必须小于或等于1小时。
为了创造一个轨迹，我们至少需要一个点。
为了将新的点与现有轨迹相关联，轨迹的最新关联点必须不比新点长1小时。
如果新的点具有完全相同的属性，但超过1小时，则需要创建新的轨迹。
我读了很多书，但是，我不能让这本书发挥应有的作用。
这是我迄今为止所做的尝试：

MATCH (p:Point)
WHERE NOT (:Trajectory)-[:CONTAINS]->(p)
WITH p.cameraSid AS cameraSid, p.trajectoryId AS trajectoryId, p.classType AS classType, p.classQual AS classQual, COLLECT(p) AS points
UNWIND points AS point
MERGE (trajectory:Trajectory{trajectoryId:point.trajectoryId, cameraSid: point.cameraSid, classType: point.classType, classQual: point.classQual, date: date(datetime(point.at))})
MERGE (trajectory)-[:CONTAINS{at:point.at}]->(point)

我不知道如何在merge子句中创建这种条件（1小时或更少）。
下面是创建一些数据的neo4j查询

// Create points
LOAD CSV FROM 'https://uca54485eb4c5d2a6869053af475.dl.dropboxusercontent.com/cd/0/get/AmR2pn0hC0c-CQW_mSS-TDqHQyi7MNVjPvqffQHhSIyMP37D7UMtfODdHDkNWi6-HqzQdp4ob2Q3326g6imEd26F3sdNJyJuAeNa8wJA2o_E6A/file?dl=1#' AS line
CREATE (:Point{trajectoryId: line[0],at: line[1],cameraSid: line[2],activity: line[3],x: line[4],atEpochMilli: line[5],y: line[6],control: line[7],classQual: line[8],classType: line[9],uniqueIdentifier: line[10]})

// Create Trajectory based on Points
MATCH (p:Point)
WHERE NOT (:Trajectory)-[:CONTAINS]->(p)
WITH p.cameraSid AS cameraSid, p.trajectoryId AS trajectoryId, p.classType AS classType, p.classQual AS classQual, COLLECT(p) AS points
UNWIND points AS point
MERGE (trajectory:Trajectory{trajectoryId:point.trajectoryId, cameraSid: point.cameraSid, classType: point.classType, classQual: point.classQual, date: date(datetime(point.at))})
MERGE (trajectory)-[:CONTAINS{at:point.at}]->(point)

如果csv文件的链接不起作用，这里有一个 alternative，在这种情况下，您必须下载该文件，然后从neo4j实例本地导入它。

Answer 1

我认为这只是其中之一，因为您可以在一个cypher语句中完成它并不意味着您应该遇到这种情况，而且您几乎肯定会发现在应用程序代码中更容易做到这一点。
无论如何，可以使用apoc并通过在轨迹节点上引入一个instanceId唯一属性来完成。
可能的解决方案
这几乎肯定不会扩展，您需要索引（稍后将根据有根据的猜测进行讨论）。
首先，我们需要更改导入脚本，以确保at属性是datetime而不仅仅是字符串（否则，我们将以datetime()调用来加速查询：

LOAD CSV FROM 'file:///export.csv' AS line
CREATE (:Point{trajectoryId: line[0], at: datetime(line[1]), cameraSid: line[2], activity: line[3],x: line[4], atEpochMilli: line[5], y: line[6], control: line[7], classQual: line[8], classType: line[9], uniqueIdentifier: line[10]})

然后，将出现以下示例数据集，并根据需要添加轨迹（并且可以在添加新点时运行）。

CALL apoc.periodic.iterate( 
'
MATCH (p: Point) 
WHERE NOT (:Trajectory)-[:CONTAINS]->(p)
RETURN p
ORDER BY p.at
',
'
OPTIONAL MATCH (t: Trajectory { trajectoryId: p.trajectoryId, cameraSid: p.cameraSid, classQual: p.classQual, classType: p.classType })-[:CONTAINS]-(trajPoint:Point)
WITH p, t, max(trajPoint.at) as maxAt, min(trajPoint.at) as minAt
WITH p, max(case when t is not null AND (
      (p.at <= datetime(maxAt) + duration({ hours: 1 })) 
      AND 
      (p.at >= datetime(minAt) - duration({ hours: 1 }))
  ) 
  THEN t.instanceId ELSE NULL END) as instanceId
MERGE (tActual: Trajectory { trajectoryId: p.trajectoryId, cameraSid: p.cameraSid, classQual: p.classQual, classType: p.classType, instanceId: COALESCE(instanceId, randomUUID()) })
ON CREATE SET tActual.date = date(datetime(p.at))
MERGE (tActual)-[:CONTAINS]->(p)
RETURN instanceId
',
{ parallel: false, batchSize: 1 })

解释
所提出的问题是棘手的，因为决定是否创建一个新的轨迹或向现有的点添加点完全取决于我们如何处理所有先前的点。这意味着两件事：
我们需要处理这些点，以确保我们创建可靠的轨迹-我们从最早的开始，然后
我们需要每一个轨迹的创建或修改都立即可见，以便处理下一个点，也就是说，我们需要孤立地处理每个点，就像它是一个小事务一样
我们将使用 apoc.periodic.iterate和 batchSize为1来给出我们需要的行为。
第一个参数构建要处理的节点集-所有当前不是轨迹一部分的点，按其时间戳排序。
apoc.periodic.iterate的第二个参数是魔法发生的地方，所以让我们把它分解-给定一个点 p到目前为止还不是轨迹的一部分：

OPTIONAL MATCH (t: Trajectory { trajectoryId: p.trajectoryId, cameraSid: p.cameraSid, classQual: p.classQual, classType: p.classType })-[:CONTAINS]-(trajPoint:Point)
WITH p, t, max(trajPoint.at) as maxAt, min(trajPoint.at) as minAt
WITH p, max(case when t is not null AND (
      (p.at <= datetime(maxAt) + duration({ hours: 1 })) 
      AND 
      (p.at >= datetime(minAt) - duration({ hours: 1 }))
  ) 
  THEN t.instanceId ELSE NULL END) as instanceId

查找与关键字段匹配且包含一个点的任何轨迹，该点位于传入点 p的一小时内，如果找到合适的匹配项，请选择其 instanceId属性（如果有多个匹配项，则为找到的最大匹配项-我们只想确保此时有零行或一行）
我们将在一分钟内看到 instanceId的全部内容，但将其视为给定 Trajectory的唯一标识符。

MERGE (tActual: Trajectory { trajectoryId: p.trajectoryId, cameraSid: p.cameraSid, classQual: p.classQual, classType: p.classType, instanceId: COALESCE(instanceId, randomUUID()) })
ON CREATE SET tActual.date = date(datetime(p.at))

确保有一个与输入点的关键字段匹配的轨迹-如果前面的代码找到了匹配的轨迹，则 MERGE没有工作要做。否则，创建新的轨迹-如果我们不较早匹配一个轨迹，我们将将属性 instanceId添加到一个新的随机UUID中，强迫 MERGE创建一个新的节点（因为没有其他的UUID存在，即使一个匹配所有其他关键字段）

MERGE (tActual)-[:CONTAINS]->(p)
RETURN instanceId

tActual现在是传入点 p应该属于的轨迹-创建 :CONTAINS关系
第三个参数至关重要：

{ parallel: false, batchSize: 1 })

重要提示：要使其正常工作，每个“inner”cypher语句的迭代都必须按顺序进行，因此我们强制将a batchSize设置为1，并禁用并行性，以防止 APOC以除一次一次之外的任何方式调度批处理。
标引
我认为，随着导入的大小和轨迹数量的增加，上面的性能将迅速下降。至少我想你需要一个综合指数
:Trajectory(trajectoryId, cameraSid, classQual, classType)-这样，找到给定点的候选轨迹的初始匹配很快
:Trajectory(trajectoryId, cameraSid, classQual, classType, instanceId) -如果结尾存在“ MERGE”，则发现现有的轨迹要快速添加，如果存在
然而，这是通过观察查询得到的猜测，不幸的是，由于我们使用的是 apoc.periodic.iterate，您无法正确地查看查询来判断执行计划是什么， EXPLAIN或 PROFILE只会告诉您有一个过程调用的命中率为1db，这是正确的，但没有帮助。