大数据环境下分布式图计算算法的改进与应用

黄承宁

(南京工业大学浦江学院，江苏 南京 211222)

0 引 言

“图计算”是以“图论”为基础的对现实世界的一种“图”结构的抽象表达，以及在这种数据结构上的数据计算模式。图数据结构很好地表达了数据之间的关联性，而关联性计算是大数据计算的核心—通过获得数据的关联性，可以从很多的海量数据中抽取有用的信息[1]。随着Web2.0、社交网络、移动互联网或者物联网的发展，数据量呈现几何级数增加，数据结构将更随意，对于信息的抽取分析也将更困难。因此，数据的增长速度也将超过计算机数据处理的增长速度，这便产生了大数据问题。为了解决在大数据环境下从海量数据中抽象出有用信息的问题，许多新型基于图数据结构的计算平台与引擎相继提出。

Google为了应对图计算的需求，推出了基于图计算的“计算框架”—Pregel，为大规模数据处理而设计的快速通用的计算引擎Spark也有支持图计算机器学习的模块—GraphX，机器学习框架GraphLab中底层框架图计算PowerGraph[2-3]。这些分布式图计算系统通过集群强大的计算资源来处理大规模的图数据，但是同步开销和容错开销使得分布式图计算系统存在瓶颈，之后GraphChi与X-Stream系统提供了解决方案，但是这些单机图计算系统在耗时性能与磁盘带宽利用率上表现不佳。GridGraph图计算系统将顶点分割成一维的chunks，再根据边的源顶点和目的顶点分割成二维的边blocks。GridGraph通过聚合更新遍历blocks一次完成最优迭代全局计算，无缝地扩展内存容量和磁盘带宽，使得GridGraph超过其他图计算系统引擎一个数量级，成为负载均衡、性能优异的图计算机系统。

1 图计算框架理论基础

在海量图计算中经常面临迭代计算，即在计算中图顶点会收到上一个循环顶点的信息，并且将这些信息汇总再次传递给下一个循环节点，在上下循环中更新节点信息。当前主流的图计算模式或者框架基本上都遵循整体BSP(bulk synchronous parallel)同步并行计算模式[4]。在此算法和框架中，每个节点会分布计算和汇总各个节点信息，区别在于不同框架中，各个节点之间存在不同的通信方式。总体框架中的一个算法由一系列SuperStep组成，如图1所示。

图1 BSP框架处理流程

BSP整个框架由一系列超步(SuperStep)组成，在这一系列步骤中，每个步骤类似节点，进行分布式计算与汇总。每个SuperStep大致可以归纳为三个环节：局部计算环节，相关资源只对本地的数据进行计算汇总；全局计算环节，相关资源只对本地之外的数据进行计算汇总；收尾计算环节，系统等待相关资源计算结束进行回收。具体如图2所示。

图2 超步三个环节

从抽象高层次看BSP是“计算-通信-同步”的模式[5-6]， 该模型包括三个模块：一组具有局部内存的分布式处理器，负责本地集群中每个节点的并行计算机进程；基于选路器(Router)的全局通信网络，用于负责在BSP模型中对图节点访问的不同Processor之间的消息数据传递，以便实现同步；支持所有处理单元间全局路障同步的机制，负责协调每个路障器的执行时序时间。即如上所述，每一个SuperStep计算过程由局部计算环节、全局计算环节和收尾计算环节组成。每一个并行完成，意味着将由本SuperStep进入下一个SuperStep。在BSP模式中最重要的是bsp方法，该方法程序代码片段如下：

public voidbsp(BSPPeer peer)throws Exception{

int in=0;

for(int i=0;i

double x=2.0*Math.random()-1.0;

double y=2.0*Math.random()-1.0;

if((Math.sqrt(x*x+y*y)<1.0)){

in++;

}

}

double data=4.0*in/iterations;

peer.send(masterTask,new DoubleWritable(data));

peer.sync();

}

其中for循环体代码完成本地分布式处理计算，peer调用send()方法完成全局不同的Processor之间的消息数据通信，peer调用send()方法实现栅栏同步。

2 并行图计算模式算法

并行处理图计算系统将图数据全部加载到集群中的内存进行计算，理论上随着集群规模的增大其计算性能和内存容量随之线性增大，能处理的图数据也随之线性增大。由于图计算理论基础BSP框架算法中存在全局通信，因此并行分布式计算模式中图分割与聚类就受到了集群网络总带宽的限制，整体性能和所能处理的图规模也存在一定的缺陷[7-9]。这类图计算系统主要包括同步计算模型的Pregel及同时支持同步和异步的系统GraphX和PowerGraph。

2.1 Pregel分布式图计算框架

Pregel提出了“像顶点一样思考”(Think Like A Vertex)的图计算模式，让用户无需考虑并行分布式计算的细节，只需要实现一个顶点更新函数，让框架在遍历顶点时进行调用即可[10]。Pregel分布式计算全过程可以描述为读取数据、初始化数据计算、开展计算和收尾计算等步骤。

读取输入初始化Pregel中的图

val originalValue=value

val value=(message:+ value).min

if(originalValue==value)

inactive()

else

neighbours.foreach(sendMessage)

运行超级步，迭代直到计算结束

var i=0

while(activeMessages>0 && i

val newVerts=g.vertices.innerJoin(messages)(vprog).cache()

g=g.outerJoinVertices(newVerts){(vid,old,newOpt).newOpt.getOrElse(old) }.cache()

messages=g.mapReduceTriplets(sendMsg,mergeMsg,Some((newVerts, activeDir))).cache()

activeMessages=messages.count()

i+=1

}

从核心代码可以看出，Pregel分布式图计算框架中每一个节点需要发送大量消息到邻居节点，同步执行容易产生滞后阻塞，从而使其在密率图的处理中性能欠佳。

2.2 PowerGraph分布式图计算框架

PowerGraph将数据抽象成Graph结构，将算法的执行过程抽象成Gather、Apply和Scatter三个步骤。其并行的核心思想是对顶点的切分。同一台机器(同一节点)上的所有edge(边)和vertex(顶点)构成Local Graph，在每台机器上，存在本地id到全局id的映射表[11-12]。

PageRank在PowerGraph中的公式：

R[i]=0.15+ΣwijR[j]

PowerGraph图计算模式：

Gather(j->i):return wji*R[j]

sum(a,b):return a+b;

Apply(i,Σ):R[i]=0.15+Σ

Scatter(i->j):if R[i] changed then trigger j to be recomp

从三个操作步骤代码可以看出，PowerGraph共享状态异步执行将需要大量锁，且每个节点的处理都触碰到了图的大部分。对于单个节点来说，元数据太大，这种边分割并行策略对于处理整个图数据依然开销太大。

2.3 GraphX分布式图计算框架

GraphX是一个新的(alpha)Spark API，用于图和并行图(graph-parallel)的计算，在Spark之上提供一栈式数据解决方案。GraphX描述的是拥有顶点属性和边属性的有向图。GraphX提供顶点(Vertex)、边(Edge)、边三元组(EdgeTriplet)三种视图。GraphX的各种图操作也是在这三种视图上完成的。GraphX将图数据以RDD分布式地存储在集群的节点上，使用顶点RDD(VertexRDD)、边RDD(EdgeRDD)存储顶点集合和边集合[13]。

图构造：

object GraphLoader {

def edgeListFile(

sc:SparkContext,

path:String,

canonicalOrientation:Boolean=false,

minEdgePartitions:Int=1)

:Graph[Int,Int]

}

图算法：

val graph=GraphLoader.edgeListFile(sc,"graphx/data/followers.txt")

val ranks=graph.pageRank(0.0001).vertices

val users=sc.textFile("graphx/data/users.txt").map {

line =>val fields=line.split(",") (fields(0).toLong, fields(1))

}

val ranksByUsername=users.join(ranks).map {

case (id,(username,rank))=>(username,rank)

}

println(ranksByUsername.collect().mkString(" "))

从算法看出GraphX是通过调用GraphLoader.edgeListFile()函数，从边文件中读入的。由于边文件中只存储了相应的顶点编号，没有顶点对应的属性。因此需要使用user(VertexId,attr)将顶点信息补全，通信开销和运行时间都比之前两者要低。

3 优化分布式图计算结合GridGraph模式实验测试比较

以上分布式图计算模式的算法过程虽然不复杂，但在实现过程中，特别在处理海量图数据时性能瓶颈变得突出，分布式的优点在于网络内的节点并行计算处理，但是在分割与聚合图过程中分布式开销太多，因此对分布式算法进行两点优化：减少分割聚合重计算过程中的内存长时间利用；减少分布式中的网络开销。

GridGraph是一个用于在单个机器上处理大规模图数据的系统。GridGraph使用预处理中的第一种细粒度级分区将图分成一维分区的顶点块chunk和二维分区的边块block。在运行时应用第二粗粒度级分区。通过一种新颖的双滑动窗口方法，GridGraph可以流化边并应用即时顶点更新，从而减少计算所需的I/O量。边的分割还使得能够进行选择性调度，可以跳过一些块以减少不必要的I/O。

GridGraph将顶点分割成P个均等的chunks。每个chunk内的顶点连续排列。全部的P×P个blocks可以看出一个网格，每个边按照如下规则被放入对应的block中：源顶点决定所在block的行，目的顶点决定所在block的列。图4展示了GridGraph如何将图3分割的例子。图3中有4个顶点，在例子中选择P=2。{1，2}和{3，4}是两个顶点chunk。例如边(3,2)被分割到Block(2,1)，因为顶点3属于Chunk 2，顶点1属于Chunk 1。

图3 图 例

图4 图分割

GridGraph只花费很短的时间就能完成完整的图分割预处理，并且同一个图生成的网格能很好地用于所有的算法。通过分割，GridGraph能进行选择调度，减少不必要的边block(没有活动顶点的边block)的访问。这对于很多迭代算法如BFS和WCC(绝大部分的顶点处于非活跃状态)在性能提升上具有极大的贡献。

实验环境采用GridGraph对比前文中分析的三种分布式图计算框架，使用i2.4xlarge实例(包含16个超线程内核，122 GB RAM，4个800 GB SSD)的GridGraph与具有16个m2.4xlarge实例(每个具有8个内核，68.4 GB RAM，2 840 GB HDD)的集群上的PowerGraph和GraphX进行性能比较，测试结果如图5所示。

图5 性能测试

实验结果表明，GridGraph无缝地扩展内存容量和磁盘带宽，单节点性能是16节点分布式系统性能的2～3倍，高出一个数量级。

4 结束语

图是一种抽象人类行为的方法，图计算的应用才刚刚开始，随着大数据研究和应用的发展，会出现更多支持“图计算”的系统。文中通过对目前流行的三种分布式图计算框架进行剖析，同时对比基于单机单节点图计算模式GridGraph的性能应用，可以看出该模式使用网格表示的大规模图表，通过分割顶点和边缘分别为1D块和2D块，能够访问存储器中的顶点数据，而不涉及I/O访问。不过该图计算模型依然受到I/O带宽的限制，框架模型还可以从网格上采用压缩技术减少来自I/O带宽的限制，进一步提升图计算性能，这些都有待进一步研究。