图计算引擎分析--GridGraph

作者：京东科技 李永萍

GridGraph：Large-Scale Graph Processing on a Single Machine Using 2-Level Hierarchical Partitioning

根据计算方法，图形计算系统可分为：单机内存图处理系统、单机核外图处理系统、分布式内存图处理系统、分布式核外图处理系统。本文将详细介绍GridGraph。

单机内存图处理系统受内存空间和单机计算能力的限制，可以解决的图规模有限。理论上，随着集群规模的增加，分布式内存图处理系统可以解决更大的图形规模，但网络带宽问题、负载不平衡、同步成本大、容错成本和图形分割挑战越来越明显。无论是单机还是分布式，内存式图处理系统能处理的图规模都是有限的。因此，如果您想使用更少的资源来解决更大的图形规模，您可以使用单机核外图处理系统。单机核外图处理系统采用磁盘顺序读写进行数据置换，可在有限的内存中计算较大的图纸。在选择调度和同步计算模式方面，单机核外图处理系统进行了重要探索，以最大限度地利用磁盘顺序读写。

GridGraph是一种单机核外图处理系统，在大规模图处理系统中充分利用磁盘读写，在有限的内存中高效地完成大规模图计算。

GridGraph充分利用磁盘大容量，解决单机内存有限时实现大规模图计算的问题。GridGraph通过Streaming-Apply减少计算中的IO 通过文件转移顺序减少不必要的io费用。 同时，GridGraph也利用顺序读写的特点，尽量少写硬盘。

GridGraph的主要贡献是：

1、基于边列表快速生成一种新的图形表示形式——网格划分。网格划分是一种不同于相邻矩阵和相邻链接表的表示形式。网格划分不需要对index进行排序。网格的边缘block可以从未排序的边缘列表转换，数据预处理成本小，可以应用于不同的算法和机器。

2、2-level hierarchical partitioning 采用两层分区划分模式，不仅适用于核外，还适用于内存。

3、为了提高IO，提出了streaming-aply模式。通过双滑动窗口（Dual sliding windows）保证顶点访问的局部性。

4、提供灵活的点边流接口函数，通过用户定制的过滤函数跳过非活动顶点(活动顶点:顶点index在bitmap中的状态为1)或非活动边计算。该方法显著提高了活动顶点集随收敛缩小的迭代算法的性能。

为了在有限的内存中完成大规模的图纸计算，并严格控制内存消耗，需要对图纸进行网格划分。

1、顶点集分为P个均匀chunk。

2、边集分为P*Pblock，行表示源顶点，列表目标顶点。

GridGraph partition预处理方法如下：

1、主线程从原始无序边集中读取边，读取一批边后，将这批边数据添加到队列中。(根据磁盘带宽，一般选择24M作为这批边的大小)

2、每个工作线程从队列中获取任务，计算边缘所属的block，并将边缘添加到边缘block文件中。为了增加I/O吞吐量，每个工作线程维护每个block的本地缓冲区，一旦缓冲区满，将刷新到文件。

GridGraph可在分区过程结束后进行计算。然而，由于现实世界图的不规则结构，一些边缘block可能太小，无法在HDD上实现大量的连续带宽。因此，由于频繁的磁盘搜索，有时无法实现顺序带宽。为了避免这种性能损失，GridGraph需要一个额外的合并阶段，以便更好地实现基于HDD的系统。在这个阶段，将边缘block文件逐一添加到大文件中，并记录元数据中每个块的起始偏移。

与Graphchi的shard分片模式不同，GridGraph不需要对边缘block进行排序，这减少了IO和计算成本。我们只需要在磁盘上读写一次边，而不是在Graphchi中多次遍历边。

对X-Stream而言，X-Stream不需要显式预处理。根据流分区，几份文件被打乱。不需要排序，分区的数量很少。只需要一个流分区，就可以将许多顶点数据安装到内存中。然而，这种分割策略使其在选择调度时效率低下，这在很大程度上影响了它在许多迭代算法中的性能，因为它只使用了某些迭代中的部分顶点。(GraphChi和X-Stream都是单机核外图计算系统，这里就不赘述了。）

调度的选择是什么？选择调度是将图数据文件(通常是边缘文件)分成多个block，并按顺序编号，设置一个bitmap来记录所有block的访问状态。如果需要访问，将bitmap中index的block编号状态设置为1，在调度过程中跳过状态为0的block，选择状态为1的block从磁盘内存中计算。如果bitmap是空的，则默认所有block都需要参与计算，然后将block按顺序从磁盘输入内存。block的大小决定了选择调度的差异。block越大，包含的数据越多，block更换的概率越低，选择调度越好。相反，block越小，包含的数据越少，计算时更换block的概率越高，选择调度越差。

GridGraph完成预处理的时间很短。此外，生成的网格格式可用于在同一图中运行的所有算法。GridGraph可以通过分区进行选择性调度，以减少不必要的访问没有活动边缘的边缘。这对许多迭代算法（如BFS和WCC）做出了巨大的贡献，因为大多数顶点在许多迭代中都不活跃。

内存（In-memory）图计算系统将所有数据读取到Memory内存中，并在系统中使用Cache(缓存)和Memory(内存)来完成图计算过程（Out-of-core）计算系统将数据存储在Disk磁盘中，然后在计算时将所需的数据替换到Memory(内存)中，通常将数据存储到Cache缓存中，以缓解CPU和Memory之间的速度差异。磁盘存储空间>存储空间>缓存存储空间。

那么如何选择Partition呢？

粒度越细(即P值越大)，预处理时间越长，P越大，每个chunk能表示的范围越广，每个block能存储的边数据越多，顶点数据访问的局部性越好，block更换概率越低，选择性调度潜力越大。因此，P越大越好。目前，我们暂时选择P的最大值，使顶点数据能够适应最后一级缓存。可以这样设置P的最小值：

(V/P)*U<=C<=>P>=C/UV

V是图中的顶点数，C是最后一级cache缓存的大小，U是每个顶点的大小。(V/P)在chunk中表示顶点范围，(V/P)*U表示每个chunk的大小。为了适应最后一级缓存，如果一个chunk的所有数据都可以一次放入最后一级缓存，那么chunk的大小应该小于或等于C，公式变换得P的最小值为C/UV.

这种分区方式不仅表现出良好的性能(尤其是在内存条件下)，而且节省了大量的预处理成本。

GridGraph采用流应用处理模型，在该模型中只需读取边缘一次，并且只需遍历一次顶点即可完成I/O总量。

GridGraph提供了两个流式处理函数，分别处理顶点(Algorithm1)和边缘(Algorithm2)：

F是一个可选的用户自定义函数，它接受顶点作为输入(Streamvertices是当前的顶点，Streamedges是block中每个边缘的源顶点)，并返回一个布尔值来指示流中是否需要顶点。当算法需要选择性的调度来跳过一些无用的流时，通常与位图一起使用，位图可以紧凑地表示活动的顶点集。

Fe和Fv是用户自定义的描述流处理函数。Fe接受一边作为输入，Fv接受顶点作为输入，返回R类型值，累积返回值，并作为最终结果提供给用户。该值通常用于获得活跃顶点的数量，但不限于此用法。例如，用户可以使用该函数来获得Pagerank中迭代之间的差异之和，以决定是否停止计算。

GridGraph将顶点数据存储在磁盘上。使用内存映射机制(通过mmap内存映射机制将顶点数据文件映射到内存)引用文件中的顶点数据，每个顶点数据文件对应一个顶点数据数组。因此，访问顶点数据文件就像访问内存中的数组一样，并简化了编程模型:开发人员可以将其视为普通数组，就像它们在内存中一样。

以Pagerank为例，我们来看看GridGraph是如何实现算法的。

Pagerank是一种链接分析算法(Algorithm3)点的数值权重，以测量其在顶点之间的相对重要性。最初所有顶点的公关值为1。在每次迭代中，每个顶点都向邻居发送自己的贡献，即当前的公关值除以其程度。每个顶点总结了邻居收集的贡献，并将其设置为一个新的公关值。当平均值差达到一定阈值时，算法收敛。

GridGraph流式读取每个block的边缘。当block在第一行j列时，与block相关的顶点数据也落在第一行j列的chunk中。每个block包含两个顶点chunk，source chunk(源顶点chunk)和destinationn chunk(目的顶点chunk)。

通过P的设置，block足够小，可以将block放入最后一级缓存中，以确保在访问与block相关的顶点数据时具有良好的局部性。

根据更新模式，block的访问顺序可以是面向行或面向列。假设顶点状态从源顶点传播到目标顶点(这是许多应用程序中的典型模式)，即读取源顶点数据，写入目标顶点数据。由于每个侧block列对应于目标顶点块，因此需要编写目标顶点块，因此在这种情况下，面向列的访问顺序优先考虑。当目的顶点的block缓存在内存中时，GridGraph从上到下流向同一列的block，因此昂贵的磁盘写作操作被聚合和最小化。特别是对于SSD系统来说，这是一个非常重要的性能，写入大量数据的性能会相应下降。另一方面，由于SSD有写入周期的上限，因此尽量减少磁盘随机写入以实现理想的持久性是非常重要的。

以Pagerank为例，我们来看看GridGraph是如何用双滑动窗口更新顶点信息的。读取窗口(从源顶点数据中读取当前顶点的Pagerank值和出度)和编写窗口(累积目标顶点新Pagerank值的贡献)作为GridGraph沿block以面向列的顺序滑动。

1、初始化，每个顶点的初始PR值为1

2、Stream edge block(1，1)此时src.chunk 1和dest.chunk 一切都加载到内存中

读取窗口：读取src.chunk 1.PR和Deg

写dest：写dest.chunk NewPR1

IO总量：读取block中的两个边，读取srcc.chunk 读取desttt1中的顶点(1，2).chunk 1中的顶点(1，2)

3、Stream edge block (2，1)此时destt.chunk 1.将src放在内存中.chunk 2也加载到内存中

读取窗口：读取src.chunk 2.PR和Deg

写dest：写dest.chunk NewPR1

IO总量：读取block中的两个边，读取srcc.chunk 2中的顶点(3，4)

4、Stream edge block (1,2)，dest.chunk 1已全部更新，更新后的destt已经完成.将srccccunk1写回磁盘种，.chunk 1和dest.chunk 2将其加载到内存中

读取窗口：读取src.chunk 1.PR和Deg

写dest：写dest.chunk NewPR2

IO总量:读取block中的两个边，destt.chunk 1中的顶点(1，2)写入磁盘，读取SRC.chunk 读取desttt1中的顶点(1，2).chunk 2中的顶点(3，4)

5、Stream edge block (2，2)此时desttt.chunk 2在内存中，src.chunk 2也加载到内存中

读取窗口：读取src.chunk 2.PR和Deg

写dest：写dest.chunk NewPR2

IO总量：读取block中的一条边，读取srcc.chunk 2中的顶点(3，4)

6、完成dest所有chunk的遍历，并完成dest.chunk 更新后的结果写入磁盘。

IO总量：desttt：.chunk 2中顶点(3，4)的结果写入磁盘。

网格图的I/O分析是在上面的一次流应用迭代中进行的，所有的边缘和顶点都被访问。以面向列的顺序访问边缘block为例:所有边缘访问一次，源顶点数据读取P次，目标顶点数据读写一次。IO用于一个完整的迭代和收敛：

E+(2+P)*V

E：读取所有边

2：读取和写入目标顶点的数据

P：读取每个P中源的顶点数据

GridGraph所需的内存非常紧凑。事实上，它只需要一个小的缓冲区来保存Stream的边缘blocl，这样页面缓存就可以使用其他空闲内存来保存更多的边缘block，当活跃的边缘block变得足够小以适合内存时，这是非常有用的。这种Streaming-Apply-Processing-Model流式应用模型的另一个优点是，它不仅支持经典的BSP模型，还允许异步更新。由于顶点更新是即时的，可以通过跟踪顶点的遍历来获得更新的效果，这使得许多迭代算法收敛得更快。可以看出，P应该是将顶点数据放入内存的最小值。因此，较小的P应该是最小化I/O量的首选，这似乎与我们上面提到的P越大越好，网格分区的原则相反。

我们之前解释过什么是选择调度，也就是跳过不活跃的边block。在Stream函数中，F传入位图跳过不活跃的边block。

P越小，粒度越粗，访问顶点的次数越少，局部性越差，调度选择越差

P越大，粒度越细，局部性越好，调度越好，访问顶点的次数越多

为了解决这个问题，二级分区应用于边网格，以减少I/O访问的顶点。

在P\*P网格中再进行一层网格划分，第二层网格有Q\*Q个边网格。在P\*P网格中应用Q\*Q分区。

应满足Q的选择：

(V/Q)*U <= M

M是给定的内存容量。

正如我们前面提到的，P的选择是将顶点数据放入容量远小于内存的上级缓存中，因此P应该远远大于Q。

整个网格分为四个大块，每个大块包含四个小块。每个块中的数字表示访问顺序。在原始的4×精确的面向列访问顺序用于4分区。二级分区应用后，P：2×2 变成 Q：4×4分区后，我们以面向列的顺序访问粗粒度(大)块，在每个大块中，我们以列为导向的顺序访问细粒度块(小)块。这种二级分区不仅提供了灵活性，还提高了效率，因为高级分区(二级分区)是虚拟分区，GridGraph可以利用低级分区(一级分区)的结果，所以不会增加更多的实际开支。并且可以使用P网格划分的结果来选择调度。

GridGraph定义了一种适应有限内存的新图表形式：网格划分；使用双窗口模式来减少IO访问的总量，特别是编写IO；选择调度来减少无用的IO；使用二级分区确保P尽可能大，减少IO访问。GridGraph在有限的内存中提高了IO效率，有效地完成了核外图的计算过程。

源码地址：https://github.com/thu-pacman/GridGraph

将原始二进制文件处理成grid格式的block文件

让我们来看看block文件是如何划分的：

将IOSIZE的数据放入buffers\[cursortasks记录了当前游标的字节数<cursor, bytes>，在tasks中获取cursor和bytes，根据cursor读取buffers中的数据，buffers[cursor根据src和dst所属的partition，将数据放入local\_buffer\[i\]\[j将local\_buffer\[i\]\[jblock\[i\]\[j在文件中。如下图所示：

代码位于：tools/preprocess.cpp

1、打开文件读取数据，并在task中添加数据。在这里，buffers的定义是全局的，tasks保存cursor和buffers的数据大小。

2、让我们来看看tasks是什么，tasks是保存当前游标和数据大小的队列。grid\_buffer\_size = 12\*8\8，12表示<4 byte source, 4 byte destination, 4 byte float typed weight>，88表示每次读取64byte的数据时，都会写一个磁盘，这是一个magicc number。

3、真正的数据处理是threads的任务。每个thread处理一个buffers\[cursor数据。

将local_buffer的数据写入相应的block文件

4、生成column文件，将所有block文件按列遍历保存到column文件中，并将每个block文件的大小保存到column_ofset文件中。

5、同理生成row文件，按行遍历读取block文件，写入row文件，并记录offset。

6、最后，将处理好的数据信息(是否包含权重、顶点数、边数、partition数)写入meta文件。

执行grid代码后，将生成P*Pblock文件、column文件、row文件column\_offset、row_offset和meta文件。

代码位于：core/graph.hpp

空间初始化，读取meta信息和column_offset、row\_offset数据，并记录每个block文件的大小

如果bitmap为空，且顶点数据字节总数(顶点数据字节总数初始化为0，则可在算法实现时设置，一般为顶点总数_顶点大小)大于0.8_内存字节数，首先获得partitionsbegin_vid和end\_vid，再次历历每一个partition，按照process执行每个partition中的每个vertex，并将返回值求和相加。最后，等待所有partition执行结束，得到begin\_vid和end\_vid。

如果bitmap不是空的，或者顶点数据字节总数小于0.8*内存字节数，则通过每个partition获得每个partition的begin_vid和end\_vid。如果bitmap是空的，它将遍历partition中的所有顶点，并根据process执行，返回值相加。否则，从begin_vid开始，根据bitmap遍历，bitmap为1的vid执行process，返回值相加。

根据bitmap决定需要遍历的partition，如果bitmap是空的，那么所有的partition都必须遍历，bitmap不是空的，根据partition是否包含bitmap中的vid，包括partition需要遍历。

统计所有需要遍历的partition文件的总大小

update默认\_mode=若update_mode=0是行更新模式(行主序更新)，update_mode=一是列更新模式(列主序)。数据准备阶段:

对于需要访问的遍历分区，分区的访问方式为:列不变，行从小到大遍历，行遍历后再向右移动。每次读取分区中IOSIZE大小的数据，最后读取PAGESIZE大小的数据，IOSIZE不够

按照process的方法执行每个侧面的操作

如果行主序，实现如下：按行遍历读取需要遍历的partition，每次处理IOSIZE大小的数据

数据处理方法是读取row文件，从offset开始读取l将ength的数据放入buffer中，然后通过process执行每个边。

让我们来看看Pagerank算法的实际使用，以Pagerank算法的实现为例，这里不再详细介绍Pagerank算法的原理。

代码位于：example/pagerank.cpp

degreeee首先初始化每个顶点：update_在这里mode=0、使用行主序更新。

每个顶点的初始pr值为1：

计算每个边的贡献值：

更新每个顶点的pr值，最后一轮迭代直接计算和更新sum：

在grid文件处理中，有几个优化点：

1）、读取输入文件时，可根据文件数量并行读取文件，加快文件处理速度。

2）、grid空间的初始化，因为每个block在初始化时不会相互影响，所以可以使用omp并行初始化来提高效率。

3）、在thread线程中，由于每个线程处理不同的cursorbuffers数据，每个thread生成自己的local_buffer写入block文件，因为threads中没有数据交互，因此也可以并行化。

stream_vertices和stream\_edges我们都进行了分析，可以看出，无论是行主序还是列主序，折线式(Z型)边block遍历策略都是必然的，其优化点如下：

1、Z型边遍历可以更改为U型遍历。以列主序为例。当遍历到最后一行SRC时，SRC保持在内存中。此时，DST向右移动，SRC从下到上遍历，以此类推，可以节省P页面替换。

GridGraph提供了一个在有限内存中完成大规模图计算的系统。解决单机内存或分布式内存无法解决的大规模图计算问题。提供一种新的切图方法，将顶点和边缘分为1D chunk和2D block表示大规模地图的网格表示；使用新的streaming-apply模型来改进IO，流化阅读边缘block的方式，并对顶点进行局部友好；在不涉及I/O访问的情况下，GridGraph可以访问内存中的顶点数据，跳过边block，不需要遍历，提高算法执行效率。

GridGraph将顶点划分为P个顶点数量相等的chunk，并将边缘放置在P*P网格中的每个block中。边缘顶点的chunk决定了网格中的行，边缘目的顶点的chunk决定了网格中的列。它对Cache/RAM/Disk采用Streamm进行了两层网格划分 vertices and 图形编程模型edges。在计算过程中，双滑动窗口（Dual Sliding Windows）I/O费用也大大降低，尤其是写作费用。以block为单位选择调度，用原子操作更新顶点，确保线程安全。为了进一步降低所需的I/O带宽，提高效率，本文提到了边网格上的压缩技术。

参考文献：

1. Xiaowei Zhu, Wentao Han and Wenguang Chen. GridGraph: Large-Scale Graph Processing on a Single Machine Using 2-Level Hierarchical Partitioning. Proceedings of the 2015 USENIX Annual Technical Conference, pages 375-386.

2. ZHU Xiaowei — GridGraph: Large-‐Scale Graph Processing on a Single Machine. Using 2-‐Level Hierarchical Parffoning. Xiaowei ZHU, Wentao HAN, Wenguang CHEN.Presented at USENIX ATC '15

3. Amitabha Roy, Ivo Mihailovic, Willy Zwaenepoel. X-Stream: Edge-centric Graph Processing using Streaming Partitions

4. Aapo Kyrola Carnegie Mellon University akyrola@cs.cmu.edu, Guy Blelloch Carnegie Mellon University guyb@cs.cmu.edu,Carlos Guestrin University of Washington guestrin@cs.washington.edu. GraphChi: Large-Scale Graph Computation on Just a PC