并行图计算: GraphX 的 pregel 接口

pregel, 是一个计算模型, 由 Google 最先提出, 后来 Spark 采用它作为迭代图计算的一个通用编程接口.

pregel 计算模型

一个 pregel 程序由一系列叫做 超步(superstep) 的迭代构成, 在每个迭代中, 每个顶点会接收到它的邻居们在上一轮迭代发送的消息, 然后改变它的顶点和边. 此外, 在每个超步结束的时候, 每个顶点也会给它的邻居们发送消息. 通过将其看作顶点，这种抽象使得对并行图处理的推理变得简单。我们所有需要关注的就是:

消息的类型, 每个顶点应该接收什么类型的消息

对进来的消息应该做什么样的处理

下一个超步中它的邻居们需要什么样的消息

幸运的是, 这种消息传递的方法有着足够的灵活性来应对一大类的图算法. 更重要的是, 一个图算法能够利用 Spark 可扩展 (scalable) 的结构以整体 (bulk) 和同步 (synchronous) 的方式来处理消息 (bulk synchronous parallel). 这种同步的计算模型能够十分容易地表达出大多数的并行图算法.

Spark 的 pregel API

GraphX 的 pregel API 定义:

class GraphOps[VD, ED] {
def pregel[A]
(initialMsg: A,
maxIter: Int = Int.MaxValue,
activeDir: EdgeDirection = EdgeDirection.Out)
(vprog: (VertexId, VD, A) => VD,
sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)],
mergeMsg: (A, A) => A)
: Graph[VD, ED]
}

pregel

方法在一个属性图上进行调用, 返回一个与原图有着同样类型和机构的新图. 当不再变化以后, 顶点的属性可能从一个超集中向下一个变化. pregel 有一下两个参数列表:

第一个列表包含:

initialMsg

: 一个为用户定义类型 A 的初始消息 - 当算法启动时该信息会被每个顶点接收到.

maxIter

: 最大迭代次数

activeDir

: 发送消息所沿边的方向

当没有消息可发送或是达到最大迭代次数, 一个 pregel 算法才会终止. 在实现算法时, 非常重要的一点是要记得设置最大迭代次数, 尤其是那些无法保证收敛的算法.

如果没有指定边的方向

activeDir

, pregel 会默认消息仅仅向每个顶点的出边发送. 此外, 如果一个顶点没有从上一个超集中接收到信息, 那么在当前超集结束的时候, 将不会向它的出边发送任何消息.

第二个参数列表必须包含三个函数:

vprog: (VertexId, VD, A) => VD

: vprog (vertex program) 会对从上一轮迭代所有接收到消息的顶点更新它们的属性

mergeMsg: (A, A) => A)

:这个函数会对每个顶点接收到的消息进行合并.

sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)]

: 这个函数接受一个 edge triplet 参数, 创建发送给边起点或终点的消息.

参考:

- Book, Apache Spark Graph Processing.