[SIGMOD 10] Pregel 基于BSP的大规模图处理系统学习总结

今天要讲的文章是SIGMOD 2010年的一篇文章，Pregel: A System for Large- Scale Graph Processing。本文主要想解决的问题就是：随着如今技术的发展，图数据规模是不断地增长的。现有的图处理系统采用单机处理大图数据，但是单机处理大图数据存在很差的可扩展性，因为单机机器内存是由限制的。然而使用MapReduce处理大图数据，效率又很差。因为每个阶段中必须存储图状态，这样在每个阶段中会造成大量的磁盘IO和网络通信带来的开销。所以本文就针对现有的图处理系统这种状况，提出了基于BSP的大规模图处理系统：Pregel。

1.Motivation

图算法常常表现出比较差的内存访问局部性，针对单个顶点的处理工作过少，以及计算过程中伴随着的并行度的改变等问题。分布式的介入更是加剧了locality的问题，并且增加了在计算过程中机器发生故障的概率。尽管大型图对象无处不在，及其在商业上的重要性，但是据我们所知，目前还不存在一种在大规模分布式环境下，可以基于各种图表示方法来实现任意图算法的，可扩展的通用系统。如图所示：

当今现有的图计算存在的问题：

1.现有的图处理系统不能满足需求重新写一个基础设施，需要大量的工程性。
2.使用MapReduce算法。没有效率，因为必须存储图状态到每个阶段中来，这样会造成大量的磁盘IO和网络通信带来的开销在每个阶段中。
3.使用单机图处理库 scalable不好
4.使用现有的并行图处理系统没有很好的容错

所以说就需要一个可以伸缩的、容错的解决方案：Pregel: A System for Large- Scale Graph Processing，并且Pregel采用一种同步批量模型。

1.Scalable and Fault-tolerant platform
2.API with flexibility to express arbitrary algorithm
3.Inspired by Valiant’s Bulk Synchronous Parallel model[4]

4.Vertex centric computation (Think like a vertex)

2.Pregel的计算模型

首先图数据从文件中输入数据，经过若干次SuperSteps后，也就是若干轮迭代之后，最终数据将会写入到文件中。每一轮迭代都叫做一个Superstep。输入数据经过compute之后进入下一轮迭代。并且每轮迭代计算都需要同步等待其他数据都计算完成后才能进入下一轮迭代。

BSP计算模型由通过通信网络连接的处理器组成。每个处理器都有一个快速的本地内存，并可能遵循不同的计算线程。 BSP计算以一系列全局超级步骤进行。一个superstep包含三个组件：
并发计算：在每个参与处理器上进行几次计算。每个进程只使用存储在处理器本地存储器中的值。这些计算是独立的，它们是异步发生的。
通信：进程之间交换数据。这种交换采取单方面的形式，并获得通话，而不是双方发送和接收电话。
屏障同步：当一个进程到达这个点（屏障），它等待，直到所有其他进程已经完成他们的通信行为。
计算和沟通行为不必按时排序。屏障同步然后结束本轮superstep。

Pregel计算和通信不必按时排序，当一个任务的计算完成之后，这个任务产生的消息就能马上发送到相应的任务中去。Pregel消息采用一种Message-Passing的方式。每个顶点接受上一个superstep产生的消息，每个顶点程序执行相同的用户顶i有的函数，然后修改它的顶点值和出边的值。接着顶点程序就会发送消息给其他顶点（这个消息会在下一个superstep中被处理）。每个顶点程序都有激活和不激活状态之分，每个顶点程序在执行一轮顶点迭代后，会经过判断是否在下一轮迭代中变成激活或者不激活状态。每个顶点程序就是一个状态机，如下图所示：

当整体图数据中，所有的顶点都变为不激活状态后，这个图迭代计算才计算完成。

3. Differences from MapReduce

Pregel
1.保持每台机器中的顶点和边进行计算
2.仅将网络传输用于消息
MapReduce
1.将图形的整个状态从一个阶段传递到下一个阶段
2.需要协调一个链接的MapReduce的步骤

4.Program API

Pregel 顶点程序定义的编程接口

Pregel 程序Example：SSSP

4. System Architecture

Pregel 采用的也是一种Master& Worker的模型。Master用来协调各个Worker工作，并且当集群发生宕机的情况下，Master还可以调度Worker，使得Worker能够从错误中恢复出来。Worker就是用来处理各个Task任务，并且每轮迭代的通信都是在不同的Worke之间进行通信。

Pregel将图数据放在分布式存储系统中（比如GFS或者BigTable），临时数据存储在本地磁盘中。

4.1Pregel执行步骤

1. Pregel初始化过程 Master将会复制很多个点程序执行到各个集群上面去

2. Master 初始化每个Worker分区的个数和Grpah分区个数，并将图graph划分到每个worker中一个或者多个分区上
3. DataInput 数据输入
Master进程为每个worker分配用户输入中的一部分，这些输入被看做是一系列记录的集合，每一条记录都包含任意数目的顶点和边。并且每个worker加载这些点集并把它标记为active 并且更新woker上的数据结构

4. Master指示每个woker执行一个superstep阶段

worker轮询在其之上的顶点，会为每个partition启动一个线程。调用每个active顶点的Compute()函数，传递给它从上一次超级步发送来的消息。消息是被异步发送的，这是为了使得计算和通信可以并行，以及进行batching，但是消息的发送会在本次superstep结束前完成。每个step会不断重复该步骤，只要有顶点还处在active状态，或者还有消息在传输。

5. 在每个点program计算都终止后，Master会指示每个worker去保存一部分执行结果

5. Fault Tolerance

5.1Checkpoints

Pregel容错采用一种Checkpoints机制，在每次进行superstep开始的时候，the master会指示每个workers去保存每个分区的状态到持久化系统中去。这个分区的状态包括：点的values 边的values 和incoming messages。然后Master自己也会保存aggregator的值。

当一个或多个worker发生故障，被分配到这些worker的partitions的当前状态信息就丢失了。Master重新分配图的partition到当前可用的worker集合上，所有的partition会从最近的某超级步S开始时写出的checkpoint中重新加载状态信息。该超级步可能比在失败的worker上最后运行的超级步 S’早好几个阶段，此时失去的几个superstep将需要被重新执行我们对checkpoint频率的选择基于某个故障模型[13]的平均时间，以平衡checkpoint的开销和恢复执行的开销。

5.2 Log outgoing Message

除了基本的checkpoint，worker同时还会将其在加载图的过程中和superstep中发送出去的消息(来自它们已经分配后的分区)写入日志。这样恢复就会被限制在丢掉的那些 partitions上。它们会首先通过checkpoint进行恢复，然后系统会通过回放来自正常的partitions的记入日志的消息以及恢复过来的partitions重新生成的消息，更新状态到S’阶段。这种方式通过只对丢失的partitions进行重新计算节省了在恢复时消耗的计算资源，同时由于每个worker只需要恢复很少的partitions，减少了恢复时的延迟。

6.Aggregator

Pregel提供一种全局通信的机制。用于全局通信，全局数据和监测，从顶点报告的值计算汇总统计信息。在superstep期间，每个工作人员从其顶点聚合值以形成部分聚合的值，在superstep结束时，来自每个worker的部分聚合值被聚合在一个树形结构中。并且树结构允许并行化，全局聚合被发送到Master节点上。Master在下一个superstep开始时将全局值发送给所有Worker。

7. Experiments