一种存储优化的分布式图处理方法

摘要

本发明公开了一种基于存储优化的分布式图处理方法，属于图计算领域。本发明包括：数据预处理阶段进行数据划分；分发图分区数据；开始数据迭代处理；更新消息传递；工作节点扩展决策；数据处理结束。本发明提出使用一致性哈希算法对图数据进行分区和存储，并设计实现了基于外存模式的分布式图处理系统，利用动态存储优化的策略，根据负载调整图的分区存储，实现图数据处理负载平衡，加快图数据处理速度，解决现有技术存在的负载不平衡，在图数据处理过程中造成热点而引起的总体性能下降问题，从而提高图处理的性能。

权利要求

1.一种存储优化的分布式图处理方法，其特征在于，包括如下步骤：

(1)初始化；将图处理系统处理机节点划分为一个主控节点和多个工作节点，各工作节点用于完成图处理的基本过程，实现图处理的计算模型；主控节点用于控制各个工作节点；

主控节点根据用户初始的配置，生成初始的消息路由表，所有工作节点保存所述消息路由表副本并与之同步更新；所述消息路由表用于记录各工作节点间的路由信息；主控节点控制整个图处理系统的执行，工作节点完成图处理的基本过程；所述路由信息，用于工作节点之间更新消息的传递和主控节点到各工作节点之间通信；

主控节点根据所述消息路由表划分图数据，按图的结点id进行划分；划分块数与消息路由表中工作节点数相同；分块根据图结点总数平均划分；所述图的所有结点形成一个环状空间，图结点id的最小值和图结点id的最大值相邻；经过划分，各图结点分区呈现两种情况，一种分区是有连续的id，另一种分区是有两段连续id；

(2)图数据的分发；主控节点将步骤(1)划分得到的各图结点分区和该分区内的元数据根据一致性哈希算法发送到消息路由表中相应的工作节点，所述元数据包括全图的边数、全图的结点数、图的类型、各分区的边数、各分区的id、各分区的起始的图结点id、各分区结束的图结点id和图分区中点图结点id；

(3)迭代次数判别；各工作节点在主控节点控制下开始图数据迭代处理；迭代前，主控节点判别迭代次数是否达到迭代次数预设值，是则转步骤(6)；否则，转步骤(4)；

(4)更新消息传递，各工作节点执行MGA计算，包括：

首先，对图分区的每条边都执行一次Map操作，每一个图结点在执行MAP操作的同时产生一个更新消息，发送给对应的目的地址；

其次，各图结点执行一个Gather操作收集传递给该图结点所有的更新消息；

第三，各图结点执行一个Apply操作，用收集的更新消息来改变这个图结点的数据；所述更新消息的发送仅仅发生在工作节点之间，更新消息根据消息路由表发送给对应的工作节点；

所述MGA计算即上述Map、Gather和Apply操作的简称，在图处理一轮迭代中，每一个图结点都要经历这三个阶段；

(5)扩展处理；主控节点根据所收集的工作节点运行状态，判别各工作节点负载是否均衡：是则，不进行分割和扩展，转步骤(3)进行下一轮迭代；

否则对负载最大工作节点的图分区数据进行分裂，即对处理的图数据进行分割，然后横向扩展，用以消除热点，即处理数据耗时最长的那个节点，采用一致性哈希算法为工作节点分配图分区数据，达到负载调控的目的；然后更新消息路由表，转步骤(3)进行下一轮迭代；

(6)图数据处理结束，同时输出计算结果。

2.如权利要求1所述的方法，其特征在于，所述步骤(4)中MGA计算过程使用图数据的流式读取保证了对存储器的顺序访问，从而保证了对外存IO的最大利用。

3.如权利要求1所述的方法，其特征在于，所述步骤(5)中收集的工作节点运行状态包括磁盘IO、网络IO以及计算消耗代价。

4.如权利要求1所述的方法，其特征在于，步骤(5)中所述热点指一轮迭代中运行最慢的工作节点；

COST＝α|V|+|E|

其中，对于图数据处理来说，α取图的平均入度，α|V|代表一个图分区所要接收的更新消息，|V|代表的是一个图分区顶点的数目，|E|代表一个图分区所要发送的更新消息，公式中COST衡量一个图的负载；分裂的目的是在需要分裂的图分区上找到一个图结点，使该图分区分裂成的两段子分区的负载代价相当。

说明书

一种存储优化的分布式图处理方法

技术领域

[0001] 本发明属于图计算领域，更具体地，涉及一种存储优化的分布式图处理方法。

背景技术

[0002] 图作为经典的数据结构，通过点和边来表达复杂的数据关系，已广泛应用于社会各领域，包括互联网领域的社交数据分析与挖掘、化学领域的蛋白质交互、医学领域疾病暴发路径的预测、学术领域中文献的引用关系等，于是衍生出很多重要的算法，包括PageRank、最短路径，连通分支，极大独立集等。正因为图数据具有重要的意义，又需要大量的计算，于是出现了各种各样的图处理系统。

[0003] 首先是分布式内存模式图处理系统，包括Pregel、GraphLab等，这些系统先把图的所有信息都放入到内存中再开始处理，这种方式执行速度快，但代价大、成本高，在规模继续增大的图应用背景下，挑战越来越显著。且单一处理机可装配内存量较为有限，处理系统横向扩展只能横向补充处理机数量，这将不可避免地增加图分区，更进一步增加切边数量，增加处理机间通信压力，加剧网络IO延迟，由此将抵消横向扩展所提供的并行优势，拖累图处理性能。

[0004] 在横向扩展遭遇矛盾时，涌现出一批采取纵向扩展设计的单机外存模式图处理技术系统，包括GraphChi、X-Stream等，其利用外存相对于内存廉价且容量更易于扩展的优点，将图的大部分数据驻留于外存，仅在计算有依赖时装载少量数据进入内存，图的信息主要通过磁盘访问的收益，减少对多机之间通信的依赖，并且可以实现在内存等资源高度受限的普通机器上进行性能可以接受的图处理，但是这种系统的性能严重受到磁盘IO的影响。

[0005] 在大数据的时代，图数据的规模越来越大，对扩展性、并行性要求越来越高。图处理系统在结构上无论是采取单机纵向扩展还是集群横向扩展均面临各自的限制。就单机而言，其资源扩展，无论是计算能力还是内存资源、IO带宽均有不足，反观分布式架构，图的合理划分早已成为经典挑战，尽管好的数据划分能平衡计算负载、减少通信开销，从而加速处理，但这种划分本身是一个NP-hard问题，即使能实现近似算法，往往也要耗费大量的时间及资源进行预处理，得不偿失，有鉴于此，现有技术仍然仅进行简单的图数据划分，如Pregel的基于hash的划分，Gemini的按段连续划分。这种简单的图数据划分在分布式图处理过程中，难以避免负载不平衡的问题，造成动态变化的处理热点，成为拖累整个图迭代处理的短板，影响图处理总体性能。

发明内容

[0006] 针对现有技术的以上缺陷，本发明提供一种存储优化的分布式图处理方法，对图数据进行分区存储和IO平衡，实现图数据处理负载平衡，加快图数据处理速度，解决现有技术存在的负载不平衡，在图数据处理过程中造成热点而引起的总体性能下降问题。

[0007] 为实现上述目的，本发明提供一种存储优化的分布式图处理方法，包括如下步骤：

[0008] (1)初始化；将图处理系统处理机节点划分为一个主控节点和多个工作节点，各工作节点用于完成图处理的基本过程，实现图处理的计算模型；主控节点用于控制各个工作节点；

[0009] 主控节点根据用户初始的配置，可以是包含各个节点信息的文件，由这个文件来生成初始的消息路由表，所有工作节点保存所述消息路由表副本并与之同步更新；所述消息路由表用于记录各工作节点间的路由信息；主控节点控制整个图处理系统的执行，工作节点完成图处理的基本过程；所述路由信息，用于工作节点之间更新消息的传递和主控节点到各工作节点之间通信；

[0010] 主控节点根据所述消息路由表划分图数据，即以图结构表示信息的大规模数据，按图的结点id进行划分；划分块数与消息路由表中工作节点数相同；分块根据图结点总数平均划分；所述图的所有结点形成一个环状空间，图结点id的最小值和图结点id的最大值相邻；经过划分，各图结点分区呈现两种情况，一种分区是有连续的id，另一种分区是有两段连续id；对图按图结点“连续”的分段，这里的“连续”就是一个图分区在上述的环状空间是连续的；这是一种基于块的分区方法，本系统初始并不对图数据进行刻意的划分，不保证分区的平衡性，而是仅仅按将图结点平均的分到每个图分区上去，而且一个图分区拥有该分区点的所有出边；

[0011] (2)图数据的分发；主控节点将步骤(1)划分得到的各图结点分区和该分区内的元数据根据一致性哈希算法发送到消息路由表中相应的工作节点，所述元数据包括全图的边数、全图的结点数、图的类型、各分区的边数、各分区的id、各分区的起始的图结点id、各分区结束的图结点id、图分区中点图结点id；后面再次分区会使用该信息；

[0012] (3)迭代次数判别；各工作节点在主控节点控制下开始图数据迭代处理；迭代前，主控节点判别迭代次数是否达到迭代次数预设值，是则转步骤(6)；否则，转步骤(4)；这里主控节点起着一个屏障的作用，屏障即图处理过程中一个工作节点要等到所有的工作节点都完成一轮迭代之后才能进行下一轮迭代；

[0013] (4)更新消息传递，各工作节点执行MGA计算，包括：

[0014] 首先，对图分区的每条边都执行一次Map操作，如PageRank算法对每一个图结点的权重按照出边的数量均分，每一个图结点在执行Map操作的同时产生一个更新消息，发送给对应的目的地址；更新消息为具体应用产生，更新消息结构包括需要传给邻结点的数据与一个目的地址，即邻接结点的地址；其次，各图结点执行一个Gather操作收集传递给该图结点所有的更新消息；第三，各图结点执行一个Apply操作，用收集的更新消息来改变这个结点的数据；所述更新消息的发送仅仅发生在工作节点之间，更新消息根据消息路由表发送给对应的工作节点；所述MGA计算即上述Map、Gather和Apply操作的简称，在图处理一轮迭代中，每一个图结点都要经历这三个阶段；

[0015] 在这一步中，各工作节点根据MGA计算模型(MGA是Map－Gather－Apply过程的简称，在图处理一轮迭代中，每一个图结点都要经历这三个阶段)，对图分区的每条边都执行一次Map操作(根据具体应用产生更新消息)，在工作节点中产生一个更新消息(更新消息结构包含一个传递给临结点的数据和一个目的地址)，发送给对应的目的地址；然后，各图结点执行一个Gather操作(收集传递给该结点的所有的更新消息)，将赋给该结点的更新消息收集，然后执行一个Apply操作(根据更新消息更新结点状态)；因此在每一轮迭代中，工作节点需要将产生的更新消息发送给对应的工作节点，这种更新消息的发送仅仅发生在工作节点之间，更新消息根据分布式消息路由表发送给对应的工作节点；所有工作节点及主控节点都要保存一份分布式系统消息路由表，其中每一个工作节点保存的分布式系统消息路由表的是主控节点所保存的分布式系统路由表的副本，当主控节点的分布式系统消息路由表更新时，主控节点会将该消息路由表同步并更新所有的工作节点消息路由表；

[0016] (5)扩展处理；主控节点根据所收集的工作节点运行状态，判别各工作节点负载是否均衡：

[0017] 是则，不进行分割和扩展，转步骤(3)进行下一轮迭代；

[0018] 否则对负载最大工作节点的图分区数据进行分裂，即对处理的图数据进行分割，然后横向扩展，用以消除热点，即处理数据耗时最长的那个节点，采用一致性哈希算法为工作节点分配图分区数据，达到负载调控的目的；然后更新消息路由表，转步骤(3)进行下一轮迭代；所述分割指将图结点分成两部分；所述横向扩展指加入一个工作节点，对分出的图数据进行处理；一致性哈希算法是为工作节点分区数据的操作，每一次添加工作节点后，都要重新分配一次；扩展是均衡负载的关键方法，也是加快一轮图处理迭代的重要手段；

[0019] (6)图数据处理结束，同时输出计算结果。

[0020] 进一步的，所述步骤(4)中MGA计算过程使用图数据的流式读取保证了对存储器的顺序访问，从而保证了对外存IO的最大利用。

[0021] 进一步的，所述步骤(5)中收集的工作节点运行状态包括磁盘IO、网络IO以及计算的消耗代价。

[0022] 进一步的，所述步骤(5)中所述热点指一轮迭代中运行最慢的工作节点，对热点的图分区进行分裂的原则是使分成的两个分区负载代价尽可能接近；

[0023] COST＝α|V|+|E|

[0024] 其中，对于图数据处理来说，α取图的平均入度，α|V|代表一个图分区所要接收的更新消息，|E|代表一个图分区所要发送的更新消息，公式中COST衡量一个图的负载；分裂的目的是在需要分裂的图分区上找到一个图结点，使该图分区分裂成的两段子分区的负载代价大致相当。该公式中COST可以理想的衡量一个图的负载。显然，扩展决策时，在需要分裂的图分区上总可以找到一个图结点，使该图分区分裂成的两段子分区的负载代价大致相当。

[0025] 本发明中，所述步骤1中划分数据使用存储优化的分区方法，这种分区策略基本思想是在初始时进行简单的按段划分，后续在图处理的过程中，系统会根据工作节点的负载情况对分区进行再次按段划分。图所有结点形成一个环状空间，图结点id的最小值和图结点id的最大值相邻，对图按图结点“连续”的分段，这里的“连续”就是一个图分区在上述的环状空间是连续的。

[0026] 本发明中，所述步骤4中MGA计算模型具体内容如图4所示，MGA过程使用图数据的流式读取保证了对磁盘的顺序访问，从而保证了对外存IO的最大利用。

[0027] 总体而言，通过本发明所构思的以上技术方案与现有技术相比，具有以下有益效果：

[0028] 本发明在处理图数据时，利用一致性哈希存储图数据，随后采取动态存储优化的策略，根据负载调整图的分区存储，从而可以实现图分区的动态扩展，消除“热点”工作节点，平衡IO，提高图处理的性能，大大提高系统处理数据的速度。

附图说明

[0029] 图1为本发明存储优化的分布式图处理系统执行流程图；

[0030] 图2为本发明存储优化的分布式图处理系统的总体架构图；

[0031] 图3为本发明存储优化的分布式图处理系统的分区策略图；

[0032] 图4为本发明存储优化的分布式图处理系统计算模型MGA示意图；

[0033] 图5、6、7为本发明存储优化的分布式图处理系统分区动态扩展示意图。

具体实施方式

[0034] 为了使本发明的目的、技术方案及优点更加清楚明白，以下结合附图及实施例，对本发明进行进一步详细说明。应当理解，此处所描述的具体实施例仅仅用以解释本发明，并不用于限定本发明。此外，下面所描述的本发明各个实施方式中所涉及到的技术特征只要彼此之间未构成冲突就可以相互组合。

[0035] 图2为本发明存储优化的分布式图处理系统的系统架构图，该系统由两部分组成，主控节点和工作节点，主控节点控制整个图处理系统的执行，工作节点完成图处理的基本过程，实现图处理的计算模型。图1为该系统数据处理的流程图，具体包括以下步骤：

[0036] 步骤1：图的划分：

[0037] 主控节点首先会根据用户初始的配置生成初始的消息路由表，根据该表划分图数据，一个分区中有两种情况，一种是有连续的id，第二种是有两段连续id，这是一种基于块的分区方法，本系统初始并不对图数据进行刻意的划分，不会保证分区的平衡性，而是仅仅按将图结点平均的分到每个图分区上去。

[0038] 步骤2：图数据的分发：

[0039] 主控节点将图分区以及分区元数据发送到对应的工作节点，包括全图的边数、全图的结点数、图的类型、分区的边数、分区的id、分区的起始的图结点id、分区的结束图结点id、图分区中点图结点id，再次分区会使用该信息；

[0040] 步骤3：开始数据迭代处理：

[0041] 主控节点控制图处理每轮迭代的开始和结束，同时会根据预设值控制迭代的次数，这里主控节点起着一个屏障的作用，屏障即图处理过程中一个工作节点要等到所有的工作节点都完成一轮迭代之后才能进行下一轮迭代。当判断全部迭代完成，执行步骤6；

[0042] 步骤4：更新消息传递：

[0043] 在每一轮迭代中，根据MGA(Map－Gather－Apply)计算模型，每一个图结点都要经历这三个阶段，图数据以流的形式从磁盘输入以便并行化处理，对每条边都执行一个Map操作，产生一个更新，这个更新结构包含一个目的地址，每个更新都会发送给对应的目的地址，每个图结点有一个Gather的操作，然后等所有的更新都收集完毕之后，每个图结点会执行一个Apply操作。工作节点需要将产生的更新消息发送给对应的工作节点，这种更新消息的发送仅仅发生在工作节点之间，更新消息根据分布式消息路由表发送给对应的工作节点；

[0044] 步骤5：扩展决策：

[0045] 扩展决策是均衡负载的关键方法，也是加快一轮图处理迭代的重要手段，主控节点会在迭代计算过程中根据主控节点所收集的工作节点运行状态，决定是否对系统进行横向扩展(scale-out)，以及对哪个工作节点进行“分裂”以消除热点，达到负载调控的目的；

[0046] 步骤6图数据处理结束，同时输出计算结果。

[0047] 本发明提供一个实施例，以[0,232-1]的环状hash空间，Vid1、Vid2、Vid3、Vid4四个图结点，3个图分区为例，其中Vid2和Vid4处于同一个分区，如图5所示，具体介绍本发明，包括以下步骤：

[0048] 步骤1图的划分：

[0049] 主控节点首先会根据用户初始的配置生成初始的消息路由表，根据该表划分图数据，一个分区中有两种情况，一种是有连续的id，第二种是有两段连续id，这是一种基于块的分区方法，本系统初始并不对图数据进行刻意的划分，不会保证分区的平衡性，而是仅仅按将图结点平均的分到每个图分区上去。

[0050] 例如图3所示，图结点是在分区时是一个环状空间，图结点首尾相接，分为四个分区，分区一、分区二、分区三中的图结点id是严格连续的，而分区四在环状空间上是连续的，跨越了图的首尾结点，实际是由两段严格连续的图结点组成。假设在分布式图处理过程中，这种基于块的方法减少了全局图结点id到本分区图结点局部id的映射开销，每个图分区只需要维护边界信息就能快速的转换。在进行负载转移的过程中，对一个图分区再次划分的时候，可以证明，沿着环查询，总可以找到一个图结点，向该图结点扩展，将该图分区再次二分，能使得这两结点的负载相等或相差最小。

[0051] 步骤2图数据的分发：

[0052] 主控节点将图分区以及分区元数据发送到对应的工作节点，首先将工作节点映射到图结点构成的环上，假设，三个工作节点通过一致性哈希Hash算法得到对应的键(KEY)值，即工作节点在这个环上的位置。

[0053] Hash(W1)＝KEY1

[0054] Hash(W2)＝KEY2

[0055] Hash(W3)＝KEY3

[0056] 然后将Key值放置于所述环中，如图6所示。

[0057] 然后采用一致性哈希算法，以顺时针的方向，将每个图分区所有的图结点(Vid)映射到离该分区最近的工作节点中(就是这个图分区归这个工作节点计算)。

[0058] 步骤3开始一轮数据迭代处理：

[0059] 主控节点控制图处理每轮迭代的开始和结束，同时会根据预设值控制迭代的次数，这里主控节点起着一个屏障的作用，屏障即图处理过程中一个工作节点要等到所有的工作节点都完成一轮迭代之后才能进行下一轮迭代。当判断全部迭代完成，执行步骤6；

[0060] 步骤4更新消息传递：

[0061] 在每一轮迭代中，根据MGA(Map－Gather－Apply)计算模型，每一个图结点都要经历这三个阶段，图数据以流的形式从磁盘输入以便并行化处理，对每条边都执行一个Map操作，产生一个更新，这个更新结构包含一个目的地址，每个更新都会发送给对应的目的地址，每个图结点有一个Gather的操作，然后等所有的更新都收集完毕之后，每个图结点会执行一个Apply操作。工作节点需要将产生的更新消息发送给对应的工作节点，这种更新消息的发送仅仅发生在工作节点之间，更新消息根据分布式消息路由表发送给对应的工作节点；

[0062] 步骤5扩展决策：

[0063] 扩展决策是均衡负载的关键方法，也是加快一轮图处理迭代的重要手段，主控节点会在每一次迭代过程中之间根据主控节点所收集的工作节点运行状态，决定是否对系统进行横向扩展(scale-out)，以及对哪个工作节点进行“分裂”以消除热点，同时采用步骤2中所描述的一致性哈希算法再次分发图数据，达到负载调控的目的。假设热点出现在第三个工作节点W3处理的分区二，则需要把分区二“分裂”成两个子分区，并添加第四个工作节点W4来处理分裂的其中一个子分区，如图7所示。

[0064] 步骤6图数据处理结束，同时输出计算结果。

[0065] 本发明在执行图数据处理时，通过对各工作节点的负载进行量化评价，使“热点”工作节点的分区数据均分，并通过一致性哈希算法来实现分区的动态扩展，对合成的图数据以及真实社交网络图数据进行测试，实验结果也很好的证明了通过存储优化，图处理的不平衡性减少，加速了图处理。

[0066] 本领域的技术人员容易理解，以上所述仅为本发明的较佳实施例而已，并不用以限制本发明，凡在本发明的精神和原则之内所作的任何修改、等同替换和改进等，均应包含在本发明的保护范围之内。