GraphGists

圣诞老人的最短加权路径

简介：圣诞老人的最短加权路径

HO! HO! HO! 今晚是平安夜，圣诞老人正乘着他的雪橇环游世界。他喜欢为每个孩子送上最好的礼物，让他们开心。从天黑起，他们就一直在等待他。他们待在壁炉旁或窗户边，抬头望着夜空，睁大眼睛捕捉哪怕是最小的飞翔驯鹿的迹象。接着，他们一个接一个地听到靠近家门口的铃声，一份新的礼物突然出现在圣诞树下，仿佛施展了魔法。圣诞老人不忙的时候，甚至可以敲开他们的家门，在送礼物之前与兴奋的孩子们面对面交流。

圣诞老人现在正在罗马的天空中。他刚刚送完城里所有的礼物，正前往下一个城镇，感到十分高兴。但是，等等！哦不，他刚刚意识到自己忘了三个孩子：Alessio、Daniela 和 Domenico！他怎么弥补这个遗漏呢？快点，圣诞老人！快点！赶在他们上床睡觉、睡着之前把礼物送给他们，否则就太晚了！！！但是怎么做呢？动动脑筋，圣诞老人！对了，尤里卡！！！他想到了！！！他立即将罗马地图建模成一个图：他认为节点可以是两条或更多街道的交叉口，而道路则是连接每个交叉口的边。关系可以携带他从一个节点到另一个节点遍历该街段所需的时间距离作为属性。将这个模型存储在像 Neo4j 这样的图数据库中，找到在每个孩子睡着之前到达他们那里的最短加权路径，成功将毫无疑问地得到保证！

reindeers pulling santas sled or sleigh 0521 1009 1013 0121 SMU

地图图模型

将街道地图表示为图是一个常见的用例场景。这可以通过多种方式实现。例如，我可以想到三种拓扑结构。

您可以将街区建模为节点，并在相邻街区之间绘制关系。

或者顶点可以代表道路，而每对相交的街道都是一条边。

第三种模型将道路交叉口视为节点。当两个或多个交叉口被一段车道直接连接，且没有其他交叉发生时，就创建一条边。以此类推。这三个示例都是正确的：您只需选择最适合您目的的模型即可。

今晚圣诞老人将使用第三种表示方法。节点有两个标签：Kid（孩子）和 RoadJunction（道路交叉口）。前者代表一个等待圣诞老人送礼物的孩子，包括他的名字和睡觉时间。为简单起见，睡觉时间表示从圣诞老人完全意识到遗漏并开始新的城市遍历到孩子睡着所需的分钟数。后者是两条或多条街道的交叉口，由其 id 唯一确定。一个节点可以同时具有这两个标签。实际上只有一种关系类型：CONNECTED_TO（连接到）。它携带属性 distance（距离），表示圣诞老人的雪橇通过该街段所需的时间。尽管它是一条有向边，但将双向遍历，假定从一侧到另一侧所需的时间与反向遍历所需的时间完全相同。

它勾勒了圣诞老人应查找最快路线的街道地图。每个交叉口由一个 id 标识，每个街段都有一个距离值。三个孩子住在 id 为 1、2、3 的道路交叉口。圣诞老人从节点 0 开始遍历。

在 Neo4j 中设置模型

让我们看看如何在 Neo4j 中将这个用例场景转换为图。共有 22 个 RoadJunction 节点。其中三个也代表孩子。他们的名字分别是 Alessio、Daniela 和 Domenico，他们的睡觉时间分别是 100、50 和 30 分钟。

为了加快查询特定 RoadJunction id 的速度，圣诞老人决定在该节点属性上创建一个索引。在这个小图中您可能无法真正体会到它的价值，但在更大的图中它至关重要。

CREATE INDEX ON :RoadJunction(id)

这里您可以看到刚刚建模的图。

MATCH (s)-[p]->(o) RETURN s,p,o;

圣诞老人的算法

圣诞老人在 RoadJunction {id:0} 顶点时意识到自己的遗漏。这将是他路线的起点。他想寻找满足以下要求的最佳路径

  • 在每个孩子睡着之前将礼物送到

  • 从起点（id 为 0 的节点）到最后访问的孩子遍历地图的旅行时间必须最短。平安夜还没有结束，还有很多孩子在其他城镇等着他们的礼物！你不想让他们错过礼物，对吧？

乍一看，圣诞老人可能会认为使用已经实现的 allShortestPaths 是个聪明的想法。然而，这个函数没有考虑每条关系的权重，即 distance 属性：实际上，更短的路径（指关系数量最少的路径）的总距离可能比更长但权重更轻的路径要高。圣诞老人需要从头开始开发一个新查询。圣诞老人的最短加权路径算法如下

MATCH (kid:Kid)
WITH count(kid) as numberOfKids
MATCH path = ((startingRoadJunction)-[:CONNECTED_TO*1..20]-(lastKid:Kid))
WHERE startingRoadJunction.id = 0 and  size(filter(x in nodes(path) WHERE (x:Kid))) >= numberOfKids
WITH  path, [x IN nodes(path) WHERE (x:Kid) | x] as kidsList, numberOfKids
UNWIND kidsList as kidInPath
WITH path, collect(distinct kidInPath) as kidsInPath, numberOfKids
WHERE  size(kidsInPath) = numberOfKids
WITH path,  REDUCE(dist = 0, rel in rels(path) | dist + rel.distance) AS pathLength, kidsInPath,
 [i in range(0,size(nodes(path))) where (nodes(path)[i]):Kid | i] as kidPositions, numberOfKids
 WHERE
  ALL (i in range (0,size(kidPositions)-1) WHERE
 reduce(dist = 0, rel in rels(path)[0..kidPositions[i]] | dist + rel.distance) < (kidsInPath[i]).bedtime)
RETURN path, pathLength
ORDER BY pathLength ASC
LIMIT 1

这里逐行解释算法。请注意，您可能需要在自己的 Neo4j 图实例中创建数据集，以避免 graphgist 平台上可能不幸出现的 org.neo4j.kernel.guard.GuardOperationsCountException 异常。

1,2) 圣诞老人首先想知道他需要送礼物给多少个孩子。这个查询很简单，可以单独执行，如下所示

MATCH (kid:Kid)
RETURN count(kid) as numberOfKids

然后通过 WITH 子句（而不是 RETURN）将结果（本例中为 3）传递给最终查询的其他部分。

3) 您正在寻找匹配此模式的子图

  • 节点 lastKid 是一个 Kid

  • 连接节点 startingRoadJunction 和 lastKid 至少需要 1 跳，最多 20 跳。这种关系的类型是 CONNECTEDTO 。为路径设置上下限总是明智的想法，特别是当您的硬件资源有限且图很大时。在本例中，20 跳足以找到路线。

  • 只要路径的最后一个节点是 lastKid，模式就是无向的。因此，关系是双向遍历的。

每个匹配此模式的路径都存储在变量 path 中。

4) 在 WHERE 条件中，设置了路径的第一个节点的 id 必须等于 0 的约束。此外

size(filter(x in nodes(path) WHERE (x:Kid))) >= numberOfKids

意味着路径中 Kid 的数量必须等于或大于先前获得的变量 numberOfKids 的值。这个条件有助于寻找最优解，可以从后续的查询步骤和计算中排除那些确定不包含所有 Kid 的路径。 filter 函数迭代路径中的所有节点，返回一个只包含标记为 Kid 的节点的数组。

5) WITH 子句允许将变量 path、变量 numberOfKids 以及包含在对应路径中找到的所有 Kid 节点（通过 filter-extract 函数）的数组传递给算法的后续步骤。

6,7) 一条路径可能多次遍历一个 Kid 节点。这两行的目的是解构每条路径的 kidsList 数组并重新收集，以便从中删除重复项。因此，结果集合是路径中包含的 Kid 的集合。

8) 另一个应用于 Kids 集合的 where 条件。它从结果中移除那些没有经过所有 Kid 的路径。您在第 4 行找到的条件不足以保证此要求，因为存在重复项。例如，一个 kidsList 为 [1,2,2] 的路径排除了 id 为 3 的 Kid，即使其大小实际上是三，也应因此被剪枝。

9,10) 此时，算法正在进行两个重要的计算，存储在变量 pathLength 和 kidPositions 中。前者是圣诞老人从开始节点到结束节点所需的总时间距离。它是通过 reduce 集合函数获得的。

REDUCE(dist = 0, rel in rels(path) | dist + rel.distance)

它迭代路径中的关系，将总距离累积到 dist 变量中，该变量初始化为 0。

后者则利用了结合 extract 和 filter 的集合函数。它返回一个数组，该数组包含 Kid 在路径所有节点中的位置。例如，假设 nodes(path) 返回的数组是 [1,4,3,4,6,2]，而 Kids 是 1,2,3。此函数将返回 [0,2,4]，即 Kids 在数组中的索引。 range 函数生成一个从 0 到路径中节点总数的数组。

[i in range(0,size(nodes(path))) where (nodes(path)[i]):Kid | i]

将此 Cypher 代码片段写成命令式编程伪代码，它将是这样

result = [];
for(int i = 0; i < nodes(path).length; i++){
  if(nodes(path)[i] is a kid)
  then result = result + i;
}
return result;

11,12,13) 这个 where 条件利用了 ALL 集合谓词：所有元素都必须符合条件。这可能是整个算法中最棘手的一步。圣诞老人为此绞尽脑汁！

ALL (i in range (0,size(kidPositions)-1) WHERE
reduce(dist = 0, rel in rels(path)[0..kidPositions[i]] | dist + rel.distance) < (kidsInPath[i]).bedtime)

它迭代包含 Kids 位置的数组。在每一步中，它检查从路径开始到第 i 个 Kid 的距离是否小于第 i 个 Kid 的睡觉时间。如果一条路径不满足此要求，它将被查询丢弃。每个子路径的距离由一个类似于上一行的 reduce 函数确定。区别在于，它不考虑路径的所有关系集合，而只考虑其切片 0..i

将此 Cypher 代码片段写成命令式编程伪代码，它将是这样

result = true;
for(int i = 0; i < size(kidPositions); i++){
    distance = 0;
    for(int j = 0; j < kidPositions[i]; j++){
        distance += rels(path)[j];
    }
    if(distance >= kidsInPath[i].bedtime){
        result = false;
    }
}

这个 ALL 谓词条件施加的约束很强，圣诞老人在选择路线时应该考虑进去。

例如，考虑以下路径，其节点 id 为 [0, 7, 1, 4, 5, 6, 9, 10, 2, 11, 15, 3]，粗体为 Kids。其距离总计 37 分钟，并在图遍历过程中到达了所有三个 Kid。然而，给 Domenico {id: 3, bedtime : 30} 送礼物时他已经睡着了，这不满足所有问题要求。

算法返回的最佳路径是 [0, 8, 9, 10, 2, 11, 15, 3, 19, 20, 14, 4, 1]。其总距离为 46.5 分钟，比前一条路径更远，需要更多的连接；然而，Kids 节点的遍历顺序使得所有孩子在收到圣诞礼物时都完全清醒。您应该注意到，Alessio，距离起点最近的 Kid，实际上是最后一个见到圣诞老人的。

结论与未来工作

在此 GraphGist 中，我们开发了一个在 Neo4j 上将街道地图建模为图的用例场景。在此基础上，使用 Cypher 构建了一个带节点约束的最短加权路径搜索算法；它利用了大多数内置的集合函数（extract、filter 和 reduce），以及 UNWIND、range 和 ALL 集合谓词。示例保持简单，以帮助读者更好地理解查询的工作原理。然而，此处阐述的原理可以轻松扩展到更大的场景。

在更大的图中测试时，我意识到一些调整可能有助于提高性能，特别是在硬件资源不足的情况下。例如，您可以在每对节点之间创建双向关系，从而将 MATCH 模式语句更改为 (startingRoadJunction)-[:CONNECTED_TO*1..20]→(lastKid:Kid)。一方面，这将增加数据库的物理大小；另一方面，它可能会提高执行查询时的整体性能。使用 allShortestPath(startingRoadJunction)-[:CONNECTED_TO*1..20]→(lastKid:Kid) 确实可以加快算法速度，但它可能会剪枝掉一些最优解，特别是在长但权重轻的路径情况下。此外，在遍历更大的数据集时，应该修改硬编码的 20 跳上限。总而言之，即使算法结果令人鼓舞，圣诞老人和孩子们因此而开心，总还是有改进的空间，尤其是在性能至关重要的生产环境中。

最后，我为您提供两个代码片段，用于生成您的随机数据集并使用查询进行尝试。

节点可以通过以下两个查询随机创建。

WITH ['Alessio', 'Daniela', 'Domenico', 'Elsa', 'Antonella', 'Alessandro',
 'Luciano', 'Susanna', 'Federica', 'Valerio', 'Fabio', 'Marta', 'Mario', 'Giustino'] as names
FOREACH (i IN range(0,100) | CREATE (kid:Kid:StreetNode {id: i, name:names[i % size(names)] + " " + i, bedtime:round(rand()*1000)}));

FOREACH (i IN range(100,1000000) | CREATE (streetNode:StreetNode {id: i}));

而边则使用这个查询。请注意，在此查询中，每个 RoadJunction 最多连接四个街段。这个假设在实际地图中也很可能发生。

WITH range(0, 1000000) as streetNodeIds
unwind streetNodeIds as streetNodeId
with streetNodeId, extract(x in range(0, toInt(round(rand()*3))) | round(rand()*10000)) as coll
unwind coll as nearStreetNodeId
match (streetNode:StreetNode {id:streetNodeId}), (nearStreetNode:StreetNode {id: nearStreetNodeId})
where streetNodeId <> nearStreetNodeId
merge (streetNode)-[r:CONNECTED_TO {distance: round(rand()*100+1)}]-(nearStreetNode)
return streetNode, nearStreetNode

感谢您读到这里！希望您喜欢！

一切顺利，

Alessio 和圣诞老人

© . All rights reserved.