CN104618153A - 分布式并行图处理中基于p2p的动态容错方法及系统 - Google Patents

Abstract

本发明涉及一种分布式并行图处理中基于P2P的动态容错方法及系统。该方法包括：定义分布式图处理问题的数据单元，以确保动态容错时所恢复数据的完整性；将处理器节点组成环状结构，将输入的图数据划分成若干分区并将各分区分派到各处理器节点中，每个处理器节点把各自的数据单元生成副本备份在邻接处理器节点中；每个处理器节点执行完自身的数据单元后，以增量的方式更新其放置于相邻处理器节点中的副本；当某个处理器节点失效或因网络出错而离线时，指派其相邻节点利用数据副本替代原先的数据单元，完成相应运算，恢复图处理的正常执行。本发明能够使图处理运算从节点失效、网络出错等异常中恢复正常状态，保证运算的正确执行。

Description

分布式并行图处理中基于P2P的动态容错方法及系统

技术领域

本发明涉及计算机网络技术领域，具体涉及一种在开放、动态的网络环境下分布式并行图处理的动态容错的方法及系统。

背景技术

近年来，随着社交网络、协同网络等技术的普及和发展，互联网中的数据规模越来越大，这给分析这些数据带来了新的挑战。在社交网络、协同网络等场景中，数据之间可能存在关联，相关研究经常用图结构来对这些数据进行描述。图中的顶点记录数据自身属性，图中的边对应数据间的关联。这样，对网络数据的分析就转化成了图处理。然而，上述图结构常常具有百万、千万个顶点和数亿条边，这对普通计算机的内存造成很大压力。更严重的是，由于图处理过程将产生正比于图规模的数量的中间结果，这使得单机由于内存的限制难以对图进行正常计算。

因此，利用分布式集群对图进行并行处理便成为了当前网络图数据分析的主要方式。以Google提出的Pregel框架为典型代表，大部分图处理系统将图数据划分成多个子图(或称图分区)分派到若干机器中，对图数据进行并行计算。并行计算主要采用BSP(Bulk SynchronousProcessing)模型，即对图数据进行迭代运算。在每个迭代步中，图中每个顶点获取邻接顶点的消息，各自更新自身状态，并将新状态传播给邻接顶点。如图1所示，在迭代步之间，所有顶点的运算需要进行同步，当所有顶点完成当前迭代步的运算时，方可触发进入下一迭代步。当所有顶点均求得结果时，图处理计算完成。

在开放的网络环境下，分布式集群具有动态变化的特点，集群中的某些节点可能失效、连接集群的网络可能出错，这些异常情况将影响图处理作业的正常执行。因此，分布式并行图处理需要考虑在运算过程中的容错性。现有的图处理系统一般通过checkpoint机制实现容错。这种容错要求图处理在发现出错时，停止当前运算，重新从磁盘中加载最近一次的checkpoint数据，从该checkpoint所记录的迭代点重新进行运算。这种容错机制是静态的，且恢复代价也比较大，并不适用于开放的网络环境。

发明内容

本发明的目的是提供一种开放、动态网络环境下分布式并行图处理中基于P2P(Peer toPeer，对等计算/对等网络)的动态容错的方法，使得图处理运算能够从节点失效、网络出错等异常中恢复正常状态，保证运算的正确执行。

为实现上述目的，本发明采用如下技术方案：

一种分布式并行图处理中基于P2P的动态容错方法，其步骤包括：

1)定义分布式图处理问题的数据单元，以确保动态容错时所恢复数据的完整性；

2)将处理器节点组成环状结构，将输入的图数据划分成若干分区，并将各分区分派到各处理器节点中，每个处理器节点把各自的数据单元生成副本备份在邻接处理器节点中；

3)在运算的每一个迭代步中，每个处理器节点执行完自身的数据单元后，以增量的方式更新其放置于相邻处理器节点中的副本；

4)当某个处理器节点失效或因网络出错而离线时，指派其相邻节点利用数据副本替代原先的数据单元，完成相应运算，从而恢复图处理的正常执行。

进一步地，步骤(1)定义的分布式图处理问题的数据单元为二元组(P_j,InMsg(P_j))，其中，P_j为由图结构划分成的某一图分区，即某一子图；InMsg(P_j)为子图P_j包含的所有顶点在某一迭代步所收到的消息集合，在运算初始时，InMsg(P_j)为空集；1≤j≤m，m为对图进行分区后分区的数量。

进一步地，步骤(2)将输入的图数据划分成若干分区后，以二元组(P_j,空的消息集合)的结构分派到各个处理器节点中。

进一步地，步骤(2)中，对于自己被分派的数据单元，处理器节点基于BSP模型对数据单元中的每一个顶点进行运算；对于相邻节点的数据单元副本，处理器节点只是记录，并等待每一次迭代步后，相邻节点发送数据来更新这些副本。

进一步地，步骤(3)在每个迭代步完成后进行副本更新；或者根据系统错误发生的频率，综合考虑副本更新所消耗的时间代价，每隔一定迭代步进行一次副本更新。

进一步地，步骤(3)在图处理的每一个迭代步中，每个顶点首先处理在上一次迭代步所收到的消息集，并根据这些消息更新自己的状态，再向邻接顶点传播自己的新状态；同一顶点在不同迭代步之间的消息不需要累积，当某一迭代步已计算出顶点的更新值后，在该迭代步之前所收到的消息都不再需要。

一种采用上述方法的分布式并行图处理系统，包括控制器和处理器节点，所述控制器负责将输入的图数据进行分区，将各分区分派到各处理器节点中，并监控各处理器节点的运行情况；各处理器节点组成环状结构，每个处理器节点把各自的数据单元生成副本备份在邻接处理器节点中，每个处理器节点执行完自身的数据单元后，以增量的方式更新其放置于相邻处理器节点中的副本；当某个处理器节点失效或因网络出错而离线时，控制器利用其相邻节点的数据副本替代原先的数据单元，恢复图处理的正常执行。

与现有技术相比，本发明的有益效果如下：

(1)传统的checkpoint机制一般需要读写磁盘，而本发明将容错时所需的副本缓存于内存中。与传统checkpoint机制相比，读写内存能更快速地实现副本记录和错误恢复；

(2)本发明基于P2P的副本分派避免了副本记录和错误恢复过程中的单一热点现象，使得分布式处理器之间的网络通信量较为平均，减少了每一个迭代步中的总体通信时间；

(3)本发明支持动态容错，即图处理在运算过程中发生异常时，可以从离异常点最近的数据单元副本恢复运算，而不必重新计算那些已经正确完成的数据，也不必从头重新执行整个图处理应用。

附图说明

图1是基于BSP模型的分布式并行图处理工作流程图。

图2是基于P2P的数据单元副本生成示意图。

图3是两个相邻处理器节点在相邻迭代步中进行副本更新的示意图。

图4是错误恢复与P2P副本的重建示意图。

图5是具体实例中UniAS架构示意图。

具体实施方式

为使本发明的上述目的、特征和优点能够更加明显易懂，下面通过具体实施例和附图，对本发明做进一步说明。

本发明的分布式并行图处理中基于P2P的动态容错的方法，包括如下步骤：

(1)为了确保在动态容错过程中能够完整地恢复数据，本发明对分布式图处理的数据单元进行了定义。

由于在图处理中，每个顶点运算所需的参数不仅包含顶点自身的原状态，还需要其邻接顶点所传播的消息。因此，在定义图处理问题的数据单元时，不仅要考虑对前一迭代的顶点中数据的依赖，还需要考虑邻接顶点的消息依赖。为了更直观地描述数据单元，本发明定义图结构如下：G＝(V,E)，其中V表示图中顶点的集合{v₁,v₂,v₃…v_n}，E表示图中的有向边组成的集合。初始的图结构划分成若干图分区组成的集合，表示为P＝{P₁,P₂,P₃…P_m}，这些分区在运行过程中将分派到各个处理器节点中。对于任意顶点v_i，其在某一迭代步所收到的消息集合为InMsg(v_i)。进一步，对于某一子图P_j，其包含的所有顶点在某一迭代步所收到的消息集合为InMsg(P_j)＝{InMsg(v_i)|all v_i in P_j}。于是，本发明定义分布式图处理的数据单元为二元组(P_j,InMsg(P_j))，其中1≤j≤m，在运算初始时，InMsg(P_j)为空集。在下文的容错技术中，要求将二元组作为一个整体看待，同时进行复制或恢复。

(2)将处理器节点组成环状结构，每个节点把各自的数据单元生成副本备份在邻接处理器节点中。

为了实现高效的容错，本发明并不将运算过程的副本保存于磁盘中，而是记录于内存中。然而，由于分布式图处理中的图规模较大，如果将每次迭代运算后的图生成副本保存在某一处理器节点中，则该节点可能成为数据传输热点，并进一步影响整体运算的效率。因此，本发明把副本的保存分散到各个处理器节点，以均衡副本产生过程中的数据传输量。本发明将参与运算过程的所有处理器节点编号，组成环状结构。根据(1)中对数据单元的定义，将输入的图数据划分成若干分区，以二元组(P_j,空的消息集合)的结构分派到各个处理器节点中。完成分派后，各个数据单元将其副本传输给编号相邻的两个处理器节点中保存。

因此，每个处理器节点既保存自己被分派的数据单元(以下称为运算单元)，也缓存着相邻节点中的数据单元副本(以下称为镜像单元)，如图2所示，其中A～H表示运算单元，A’～H’表示镜像单元。对于自己被分派的数据单元，处理器将基于BSP模型对数据单元中的每一个顶点进行运算；对于相邻节点的数据单元副本，处理器只是记录，并等待每一次迭代步后，相邻节点发送数据来更新这些副本。

(3)在运算中，每个处理器节点执行完自身的数据单元后，以增量的方式更新其放置于相邻处理器节点中的副本。

在分布式图处理系统中，图的每个顶点是随着迭代运算不断更新的，这使得步骤(2)中所设置的副本需要不断更新。而由于每个顶点的运算需要其相邻顶点在上一迭代步所传播的消息，这使得在数据恢复过程中必须考虑多个数据单元副本所处的迭代步之间的关系问题。

为了简化这一问题，保证副本的更新和副本所处迭代步的一致。本发明基于BSP计算模型，在两个迭代步之间增添了副本更新阶段，在该阶段，要求图处理中的所有数据单元都要根据当前值更新其副本。副本更新阶段可以发生在每个迭代步完成后，也可以根据系统错误发生的频率，综合考虑副本更新所消耗的时间代价，每隔一定迭代步进行一次副本更新。

为了降低每次副本传输所消耗的时间与网络资源代价，并减少副本缓存所占用的内存空间，本发明提出了增量式的副本更新。每次对副本中图分区状态的更新只传输发生变化的部分，相邻处理器节点将变化的部分累加到数据单元副本中，实现副本中图分区状态的更新。此外，由于在图处理的每一个迭代步中，每个顶点将执行如下操作，首先处理在上一次迭代步所收到的消息集，并根据这些消息更新自己的状态，再向邻接顶点传播自己的新状态。因此，同一顶点在不同迭代步之间的消息是不需要累积的，当某一迭代步已计算出顶点的更新值后，在该迭代步之前所收到的消息都不再需要。以某个图分区P_j为例，在初始时将(P_j ⁰,null)生成副本缓存在相邻处理器节点中。在第1个迭代步运算完成时，该分区的状态更新为P_j ¹，令分区增量△P¹为P_j ¹与P_j ⁰的差值。则P_j所在的处理器单元将(△P¹,InMsg(P_j)[1])发送给相邻的处理器节点。相邻处理器节点将△P¹累积到副本单元(P_j ⁰,null)中生成P_j ¹，并替换消息集合为InMsg(P_j)[1]。以此类推，在第k个迭代步运算完成时，令分区增量△P^k为P_j ^k与P_j ^k-1的差值。在副本更新阶段，将(△P^k,InMsg(P_j)[k])发送给相邻处理器节点。相邻处理器节点将△P^k累积到副本单元(P_j ^k-1,InMsg(P_j)[k-1])中生成P_j ^k，并删除InMsg(P_j)[k-1]，替换消息集合为InMsg(P_j)[k]。图3展示了两个相邻处理器节点在相邻迭代步中进行副本更新的过程。

(4)当某个处理器节点失效或因网络出错而离线时，指派其相邻节点利用数据副本替代原先的数据单元，完成相应运算，从而恢复图处理的正常执行。

当图处理运行到某一迭代步k时，如果某个处理器节点失效或因网络出错而离线，在该离线节点中的数据单元无法继续运算，但它们的副本(P_j,InMsg(P_j))缓存于相邻节点中。基于步骤(3)所提到的副本数据增量保存的方式，本发明可以从相邻节点的副本中恢复出这些丢失的数据单元，如图4所示。

为此，本发明设计了如下动态容错算法：

Recovery_scheduling(离线节点ID，离线节点中的数据单元集合Set){

获取离线节点的两个相邻节点L,R；

if(L在线&&R在线){

根据节点L,R的负载情况，将Set划分成两个子集SetL,SetR；

节点L将自己缓存中与SetL相对应的镜像单元设置成运算单元，并完成这些SetL中的所有顶点在当前迭代步的运算；

节点R将自己缓存中与SetR相对应的镜像单元设置成运算单元，并完成这些SetR中的所有顶点在当前迭代步的运算；

设置节点L,R为相邻节点；

对比节点L,R中的运算单元，若在对方节点中不存在镜像单元的，生成该数据单元的副本发送到对方节点中；

恢复正常执行；

}else if(L在线||R在线){

N＝在线的节点；

节点N将自己缓存中与Set相对应的镜像单元设置成运算单元，并完成这些Set中的所有顶点在当前迭代步的运算；

寻找N的相邻节点；

对比节点N与其相邻节点中的运算单元，若在对方节点中不存在镜像单元的，生成该数据单元的副本发送到对方节点中；

恢复正常执行；

}

}

由于上述方案是对原有BSP模型的扩展，所以并不影响图处理系统在正常环境下的执行。而利用本发明所提供的方案，图处理系统的容错并不需要停止当前运算，全部回滚到最近的checkpoint点，从磁盘中读取副本数据。在处理器节点离线时，图处理系统只需要对出错的局部节点进行调整恢复，而其他正常节点所完成的运算结果可以继续用到下一个迭代步中。

至此，本发明完成了分布式并行图处理中的动态容错。

以下给出一个在分布式并行运算平台UniAS上使用本发明构建支持动态容错的分布式并行图处理的实施案例。

UniAS是由北京大学信息科学技术学院软件研究所自主研发的分布式并行运算平台，目前已支持包括批处理、图处理和流处理等多种模式的大数据处理应用。下面围绕UniAS中的图处理框架介绍本发明的实施过程。

如图5所示，UniAS中的图处理框架基于Master-Slave结构实现。控制器负责将图进行分区，并将各分区分派到处理器节点中，并监控各个节点的运行情况。每个处理器节点均有一个数据单元队列模块，负责运行过程中的副本维护和动态容错。根据本发明的技术要点，下文通过若干步骤实现动态容错的分布式图处理：

1.启动处理器节点，向控制器注册。控制器将上线的所有处理器节点进行编号，排列成环状结构；

2.初始化阶段，控制器将输入的图结构进行分区，并将各分区分派到处理器节点中。分派完成后，触发各节点的副本生成操作，各节点将自身的数据单元制作副本发送到相邻节点中；

3.启动图处理运算。根据本发明技术点3(即上述步骤(3))的要求，在每个迭代步完成时，求出每个顶点上一迭代步与当前迭代步结果的差值。若顶点状态有改变，则将差值加入分区增量中；若顶点在本次迭代中没有发生修改，则不加入分区增量。将每个分区的分区增量与消息集合发送给相邻节点，用于相邻节点的副本更新；

4.每个处理器节点定期向控制器发送心跳消息。当控制器未能及时收到某个处理器节点的心跳消息时，控制器确认该处理器节点离线，进入错误恢复阶段；

5.控制器调用本发明技术点4(即上述步骤(4))所设计的算法，利用相邻处理器的副本恢复丢失的数据，并重新设置处理器节点间的邻接状态，从而恢复程序的正常运行；

6.当在某一个迭代步，所有顶点都处理完成，所有数据单元都不需要再运算更新时，整个图处理作业成功完成。

至此，利用本发明在UniAS平台上构建了分布式图处理框架的动态容错机制。

以上实施例仅用以说明本发明的技术方案而非对其进行限制，本领域的普通技术人员可以对本发明的技术方案进行修改或者等同替换，而不脱离本发明的精神和范围，本发明的保护范围应以权利要求书所述为准。

Claims (8)

1.一种分布式并行图处理中基于P2P的动态容错方法，其步骤包括：

1)定义分布式图处理问题的数据单元，以确保动态容错时所恢复数据的完整性；

2)将处理器节点组成环状结构，将输入的图数据划分成若干分区，并将各分区分派到各处理器节点中，每个处理器节点把各自的数据单元生成副本备份在邻接处理器节点中；

3)在运算的每一个迭代步中，每个处理器节点执行完自身的数据单元后，以增量的方式更新其放置于相邻处理器节点中的副本；

4)当某个处理器节点失效或因网络出错而离线时，指派其相邻节点利用数据副本替代原先的数据单元，完成相应运算，从而恢复图处理的正常执行。

2.如权利要求1所述的方法，其特征在于：步骤(1)定义的分布式图处理问题的数据单元为二元组(P_j,InMsg(P_j))，其中，P_j为由图结构划分成的某一图分区，即某一子图；InMsg(P_j)为子图P_j包含的所有顶点在某一迭代步所收到的消息集合，在运算初始时，InMsg(P_j)为空集；1≤j≤m，m为对图进行分区后分区的数量。

3.如权利要求2所述的方法，其特征在于：步骤(2)将输入的图数据划分成若干分区后，以二元组(P_j,空的消息集合)的结构分派到各个处理器节点中。

4.如权利要求2所述的方法，其特征在于：步骤(2)中，对于自己被分派的数据单元，处理器节点基于BSP模型对数据单元中的每一个顶点进行运算；对于相邻节点的数据单元副本，处理器节点只是记录，并等待每一次迭代步后，相邻节点发送数据来更新这些副本。

5.如权利要求1所述的方法，其特征在于：步骤(3)在每个迭代步完成后进行副本更新；或者根据系统错误发生的频率，综合考虑副本更新所消耗的时间代价，每隔一定迭代步进行一次副本更新。

6.如权利要求1所述的方法，其特征在于：步骤(3)在图处理的每一个迭代步中，每个顶点首先处理在上一次迭代步所收到的消息集，并根据这些消息更新自己的状态，再向邻接顶点传播自己的新状态；同一顶点在不同迭代步之间的消息不需要累积，当某一迭代步已计算出顶点的更新值后，在该迭代步之前所收到的消息都不再需要。

7.一种采用权利要求1所述方法的分布式并行图处理系统，其特征在于，包括控制器和处理器节点，所述控制器负责将输入的图数据进行分区，将各分区分派到各处理器节点中，并监控各处理器节点的运行情况；各处理器节点组成环状结构，每个处理器节点把各自的数据单元生成副本备份在邻接处理器节点中，每个处理器节点执行完自身的数据单元后，以增量的方式更新其放置于相邻处理器节点中的副本；当某个处理器节点失效或因网络出错而离线时，控制器利用其相邻节点的数据副本替代原先的数据单元，恢复图处理的正常执行。

8.如权利要求7所述的系统，其特征在于：每个处理器节点定期向控制器发送心跳消息，当控制器未能及时收到某个处理器节点的心跳消息时，控制器确认该处理器节点离线，进入错误恢复阶段。