Spark的过滤操作如何在GraphX边缘上工作？

Question

我对Spark很陌生，不太了解基础知识，我只是为了解决一个问题而跳入其中。该问题的解决方案包括制作一个边具有字符串属性的图（使用GraphX）。用户可能希望查询这个图，我通过只过滤那些具有字符串属性的边来处理查询，该属性等于用户的查询。

现在，我的图形有超过1600万条边；当我使用计算机的所有8个核心时，创建图形需要10分钟以上。然而，当我查询这个图时（就像我上面提到的那样），我会立即得到结果（令我惊喜）。

所以，我的问题是，过滤器操作到底是如何搜索我所查询的边缘的？它是否迭代地查看它们？是在多个核心上搜索边缘，而且看起来很快吗？还是涉及到某种散列？

下面是如何使用filter:mygraph.edges.filter(_.attr(0).equals（“cat”））的一个示例，这意味着我希望检索具有属性“cat”的边。边缘搜索得怎么样了？

Answer 1

运行语句返回速度如此之快，因为它实际上并不执行筛选。Spark使用惰性求值：直到您执行一个实际收集结果的操作，它才真正执行转换。调用转换方法（如filter）只需创建一个表示此转换及其结果的新RDD。您必须执行诸如collect或count之类的操作才能实际执行：

scala prettyprint-override">def myGraph: Graph = ???

// No filtering actually happens yet here, the results aren't needed yet so Spark is lazy and doesn't do anything
val filteredEdges = myGraph.edges.filter()

// Counting how many edges are left requires the results to actually be instantiated, so this fires off the actual filtering
println(filteredEdges.count)

// Actually gathering all results also requires the filtering to be done
val collectedFilteredEdges = filteredEdges.collect

请注意，在这些示例中，筛选结果不存储在两者之间：由于惰性，对两个操作都重复筛选。为了防止这种重复，在阅读了转换和操作的详细信息以及Spark在幕后的实际操作之后，您应该研究Spark的缓存功能:https://Spark.apache.org/docs/latest/programming-guide.html#rdd-operations。

在Spark GraphX中，边存储在edgerdd[ED]类型的RDD中，其中ED是边缘属性的类型，在您的例子中是string。这个特殊的RDD在后台进行了一些特殊的优化，但出于您的目的，它的行为类似于它的超类RDD[edge[ED]]，筛选类似于筛选任何RDD：它将遍历所有项，并对每个项应用给定的谓词。然而，一个RDD被拆分为许多分区，Spark将并行过滤多个分区；在您的示例中，如果您似乎在本地运行Spark，它将并行执行与您拥有的内核数量一样多的操作，或者与您用-例如master local[4]显式指定的内核数量一样多的操作。

带边的RDD根据设置的PartitionStrategy进行分区，例如，如果您使用graph.FromedGetUples创建图形，或者通过调用图形的PartitionBy来创建图形。然而，所有策略都基于边缘的顶点，因此不需要了解您的属性，因此不会影响您的过滤操作，除非在集群上运行它，否则可能会出现一些不平衡的网络负载，所有“cat”边缘最终会出现在同一个分区/执行器中，然后执行collect或一些shuffle操作。有关如何表示和划分图的更多信息，请参见关于顶点和边RDDs的GraphX文档。