并行图计算: GraphX 的 pregel 接口

雨点打透心脏的1/2处 2022-06-12 00:06 151阅读 0赞

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.

[bulk synchronous parallel]: https://en.wikipedia.org/wiki/Bulk_synchronous_parallel