Spark GraphX: 改变图的结构

喜欢ヅ旅行 2022-06-12 01:14 195阅读 0赞

Spark GraphX 提供了 4 个十分有用的方法来改变图的结构， 方法签名如下:

class Graph[VD, ED] { 
         def reverse: Graph[VD, ED]
         def subgraph(epred: EdgeTriplet[VD,ED] => Boolean,
                      vpred: (VertexId, VD) => Boolean): Graph[VD, ED]
         def mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED]
         def groupEdges(merge: (ED, ED) => ED): Graph[VD,ED]
       }

### reverse ###

def reverse: Graph[VD, ED]

正如字面意思 (reverse 有逆向, 反转的意思), `reverse` 操作符将会返回一个所有边的方向都反转的新图. 此外, 在实现上, 它也不会产生任何数据移动或复制.

### subgraph ###

def subgraph(epred: EdgeTriplet[VD,ED] => Boolean,
                      vpred: (VertexId, VD) => Boolean): Graph[VD, ED]

`subgraph` 对过滤一个图十分有用. 它接收两个返回值为布尔类型的判定函数, 第一个判定函数 `epred` 接受一个 EdgeTriplet 参数, 当 triplet 满足判定函数时返回 true. 同样, `vpred` 接受一个 (VertexId, VD), 当顶点满足判定函数时返回 true.

通过这两个判定函数, `subgraph` 所返回的图中仅包含满足条件的顶点和边. 默认情况下, 如果没有提供判定函数, 那么会直接返回 true. 这意味着我们仅能传入一个 edge 的判定函数, 一个 vertex 的判定函数, 或者两个同时传入. 如果 `subgraph` 仅传入了一个 vertex 的判定函数并且两个相邻的节点被过滤了出去, 那么这些节点所连接的边也会被自动过滤.

`subgraph` 操作符对于很多场景都非常方便. 比如, 在实际应用中, 有些图有一些独立的顶点或是缺失顶点信息的边. 我们就可以使用 `subgraph` 过滤掉这些元素. 另一个场景是当我们想要去除图中的 “超级节点”, 也就是那些与很多节点都相连的大型节点.

下面举一个具体的例子, 用 `subgraph` 来解决在社交网络中经常遇到的一个问题: “在我的朋友的朋友中, 有多少还不是我的朋友?”:

// Given a social network
       type Name = String
       class Person(name: Name, friends: List[Name])
       val socialNetwork: Graph[Person, Int] = ...
       // that I am part of
       val me = Person(myName, myFriends)
       // I want know my friends' friends that are not yet my friends
       val potentialFriends = socialNetwork.subgraph(vpred = 
       (_, p: Person) => !(me.friends contains p.name))

### mask ###

def mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED]

`mask` 也是对一个图进行过滤. 与 `subgraph` 不同的是, `mask` 并不接受一个判定函数作为参数. 相反, 它的参数是另一个图. 然后, `graph.mask(anotherGraph)` 会构造出一个 `graph` 的子图, 这个所包含的顶点和边须同时在 `anotherGraph` 中出现. 基于另一个相关图的属性, `mask` 可以和 `subgraph` 一起用来完成对一个图的过滤.

考虑如下场景, 我们想要找出一个图中的连通组件, 但是同时想要移除在结果图中缺失属性的顶点. 我们可以使用 `connectedComponents` 算法, 然后在使用 `mask` 和 `subgraph` 来获得最终结果:

// Run Connected Components
       val ccGraph = graph.connectedComponents()
       // Remove vertices with missing attribute values and the edges connected to them
       val validGraph = graph.subgraph(vpred = (_, attr) => attr.info != "NA")
       // Restrict the resulting components to the valid subgraph
       val validCCGraph = ccGraph.mask(validGraph)

### groupEdges ###

def groupEdges(merge: (ED, ED) => ED): Graph[VD,ED]

`groupEdges` 可以将两个节点间的平行边变为单独的一条边. 为此, `groupEdges` 需要一个叫做 `merge` 的函数作为参数, `merge` 接受一个 (ED, ED) 的 pair 作为参数, 并将这两个 ED 合并为同一类型的单个属性. 因此, `groupEdges` 所返回的图与原图具有同样的类型.

参考:

*  Book, Apache Spark GraphX Processing