问题标签 [pregel]

'

我正在通过 Graphx In Action 工作，这本书（源代码在这里：https ://github.com/insidectm/spark-graphx-in-action ）讨论了两种计算距离的方法（边缘跳数) 在树的根和所有节点到叶子之间。我了解使用 aggregateMessages 提供的代码示例。特别是，停止条件是有意义的（我通过包含文本“停止条件”的注释突出显示了该条件，如下。）一旦图形顶点上的属性停止变化，继续运行算法。

当我查看 Pregel 计算相同结果的方式时，我有点困惑（如下所示。）

特别是在调用 Pregel 的 apply 方法时，maxIterations 是默认值，即 Integer.MAX_VALUE（出于所有实际目的，“永远运行”。）因此，“sendMsg”函数似乎是：

将被无限调用，即使在顶点上的值已经收敛之后。

是否有一些我忽略的机制导致程序在收敛后停止？

我正在使用来自 jupyter 的 pyspark 和 graphframes。我能够成功导入 pyspark 和 graphframes，但是当我尝试时：

我收到以下错误：

这篇文章是我如何让 graphframes 工作，但没有 graphframes.lib：

https://github.com/graphframes/graphframes/issues/104

我尝试重复上述命令，没有环境行，因为 pyspark 在 jupyter 中对我来说工作得很好，使用不同的版本并且能够获取 graphframes.lib，但没有 Pregel：

现在我可以看到graphrames.lib 目录，但里面只有aggregate_messages.py。

最后，我尝试了以下但收到 404 错误：

我希望，因为我能够导入图形框架，所以我能够从 graphframes.lib 导入 Pregel。似乎在我的版本中，现在是 0.6.0，有一个 graphrames.lib 但没有 Pregel，并且还没有用于 graphframes 的 0.7.0 版本。

我正在尝试使用 Pyspark 和来自 graphframes 的 pregel 包装器来实现 Rocha & Thatte ( http://cdsid.org.br/sbpo2015/wp-content/uploads/2015/08/142825.pdf ) 的算法。在这里，我遇到了消息聚合的正确语法。

这个想法是向前发展的：

...在每次传递中，G 的每个活动顶点都将一组顶点序列发送到其邻居，如下所述。在第一遍中，每个顶点 v 将消息 (v) 发送给它的所有邻居。在随后的迭代中，每个活动顶点 v 将 v 附加到它在先前迭代中接收到的每个序列。然后它将所有更新的序列发送给它的邻居。如果 v 在上一次迭代中没有收到任何消息，则 v 将自己停用。当所有顶点都被停用时，算法终止。...

我的想法是将顶点 ID 发送到目标顶点（dst），并在聚合函数中将它们收集到一个列表中。然后在我的顶点列“序列”中，我想将这个新列表项与现有列表项追加/合并，然后使用 when 语句检查当前顶点 ID 是否已经在序列中。然后我可以根据顶点列将顶点设置为 true 以将它们标记为循环。但是我在 Spark 中找不到关于如何连接它的正确语法。有人有想法吗？或者实现了类似的东西？

我当前的代码

代码不起作用，但我认为逻辑应该是

我在 Spark 中使用 Pregel 编写了处理图形的代码，但是对于一个小数据集，它的执行速度非常非常慢。我以前用 pregel 写过程序，但是这段代码运行起来很慢。我的集群由 2 个工作人员组成。每个都有核心 i5 CPU 和 6 GB 的 RAM。这是我在 Pregel 中编写的代码：

谁能解释这个执行缓慢的问题在哪里？是不是因为我有 2 个 worker 并且 Pregel 中有很多消息传递和归约操作，所以我的代码在大数据集中的性能下降了很多？

我尝试为一个相对较小的图（250k 顶点，1.5M 边）运行我自己的 Pregel 方法。在大多数情况下，我使用的算法可能（很有可能）是非收敛的，maxIterations 设置实际上是硬停止完成所有计算。

我将 AWS EMR 与 apache spark 和 m5.2xlarge 实例一起用于具有 EMR 托管扩展的设置中的所有节点。最初，集群设置为运行 1 个主节点和 4 个工作节点，最多可扩展至 8 个。

对于相同的集群设置，我将 maxIterations 的数量从 100 逐渐增加到 500，步长为 100 [100, 200, 300, 400, 500]。我假设设置足够 100 次迭代对于任何其他数字也足够了，因为未使用的内存将被释放。

但是，当我运行一组 maxIterations 从 100 增加到 500 的作业时，我发现 maxIterations > 100 的所有作业都由于步骤错误而终止。我检查了 Spark 的日志以发现问题，这就是我得到的：

日志开始

然后大约 2Mbytes 的相同输出，然后它完成：

• 由于某些集群节点上的 OutOfMemory 错误，Pregel 没有完成 200 次或更多次迭代，我是否正确？
• 如果是这样，Pregel 如何在 100 次迭代没有导致它并且 200 或 300 次导致它的情况下工作？在此问题之前，我的理解是 Pregel 与许多其他迭代方法一样，仅“存储”先前和当前的迭代值，结果和迭代值的迭代正在发生变化，但它们的数量并没有增加，这意味着它仍然是具有 250k 顶点和 1.5m 的图边缘和只有对当前迭代有效的消息才会加到堆中。
• 在整个日志中，我无法找到有关内存不足的任何信息，并且如所见，每个节点在终止之前都有千兆字节的可用内存

我正在尝试使用图形框架和 Pregel API 构建一棵树。

我有以下联系：

并需要以下输出作为数据框：

实际代码如下所示：

我不知道为什么，但 dst_id 附加没有按预期工作。我尝试了许多不同的方法，但在使用列表列表时仍然出现相同的错误。