[论文笔记]Maiter：一种基于Delta的累积迭代计算的异步图处理框架

Maiter: An Asynchronous Graph Processing Framework for Delta-based Accumulative Iterative Computation

背景：

传统的分布式图处理框架一般为同步迭代的方式，也就是同一轮迭代内并行；不同轮迭代间串行。典型的同步分布式图计算框架包括Pregel和MapReduce。同步迭代方式最大的局限在于水桶效应，执行速度最慢的worker会影响整个系统的执行效率，因为只有所有worker都完成了本轮迭代计算，系统才能进入下一轮迭代。
据此人们提出了一些异步分布式迭代计算的框架，然而这些框架存在着相当严重的问题：通信量增加、存在大量无效计算。比如PageRank算法，顶点根据所有邻居顶点的权值更新自身的权值，然而如果某个邻居顶点的权值没有更新，那么该顶点本次计算的权值就是无效的。同时每个顶点频繁地与邻居通信，会造成通信负载过大。
除此之外，异步迭代不保证能获得与同步迭代相同的计算结果，存在不确定性。
作者在本篇文章中提出了一种新型的异步分布式迭代计算框架DAIC，并实现了其对应的分布式计算系统Maiter。

DAIC介绍

DAIC-Delta-based Accumulative Iterative Computation，是本文提出的一种异步分布式迭代计算模型。各顶点的单次迭代过程可分为两步：

然而不是所有图迭代算法都能使用DAIC模型，该算法必须满足以下4个条件：

（1）

也就是顶点 j 在第k次迭代的状态，可以根据其入边邻居顶点的状态得到。比如PageRank算法，就满足上述条件。

（2）
即运算符⊕必须满足分配性。

（3）
运算符⊕必须满足交换律和结合律

（4）
根据上述四个条件和初始的两步迭代等式，我们可以根据所有顶点的初始值v_j⁰和Δv_j¹求出任意顶点在迭代次数k下的值v_j^k。

然而上述模式仍然是同步的，即：

所以上述的模式称为同步DAIC，接下来介绍异步DAIC模式。

异步DAIC
DAIC可以是异步迭代的，即顶点可以基于其获得的入边邻居顶点的信息，在任意时刻进行状态update操作。异步DAIC模式可表示为两步操作：

其中 m_j 是顶点 j 的任意邻居顶点发来的信息。每个顶点设置有一块缓存，用来保存这些mj，作为Δv_j，用作更新值。
当顶点v_j判断其缓存的信息包含了所有入边邻居顶点在状态k下的信息时，即可执行更新操作，进入状态k+1，发送权值给出边邻居顶点，并将缓存Δv_j清零。

收敛性
作者证明了当迭代次数趋于无穷时，异步DAIC能够得到与同步DAIC同样的迭代结果。

有效性
作者证明了异步DAIC的收敛速度比同步DAIC更快。

调度策略
对于图的分布式并行计算，一般是将图进行点中心或边中心分割，交给多个处理器进行计算，每个处理器负责一个子图的计算。
在异步DAIC中，单个处理器对子图更新时，有两种策略：循环调度和优先级调度。循环调度：以循环的方式更新子图中的顶点；优先级调度：优先更新子图中权值变化最大的顶点。
作者证明了优先级调度的异步DAIC能获得与同步DAIC相同的迭代结果。

作者还证明了异步DAIC的优先级调度策略比循环调度策略收敛得更快。

DAIC算法

作者在本章节给出了若干迭代算法的DAIC版本。包括SSSP、Connected Components、PageRank、Adsorption、HITS、Katz metric、Jacobi method、SimRank、Rooted PageRank。这些算法都满足上述的4个条件，其在DAIC模式下的符号如下表所示。

Maiter介绍

Maiter是根据异步DAIC模型实现的分布式异步计算系统。用户将算法转换为DAIC算法时，只需实现Maiter提供的几个API，指定上述Table 1 中的四个数值或函数，以及图分区，迭代终止条件等信息。
Maiter系统拥有一个master和其余若干worker，master协同worker，监督worker的工作状况。worker之间并行执行，并使用MPI进行信息交流。
下面是一些Maiter的实现方式：

Data Partition：图中每个顶点拥有独一无二的vid，Maiter根据hash函数h()对顶点进行分区。

Local State Table：每个worker维护一张本地状态表，记录本地顶点的状态。如下图所示。

Receive Thread and Update Thread：接收线程-接收入边顶点传来的信息，更新本地状态表；更新线程-更新本地状态表，发送信息给出边顶点。

Scheduling within Update Thread：Maiter使用优先级更新策略，在一轮更新过程中，Maiter根据vid优先级向更新队列中装入若干待更新顶点，依次更新顶点并将其出队。当更新队列为空时，再次根据优先级挑选若干待更新顶点，进入下一轮更新。

Message Passing：Maiter维护了多张发送表-msg tables，每张表对应一个worker。当Update线程执行后，会产生若干message，这些message会根据目的地装入对应的msg table中，（注意，这里的装入不是作为一个新的数据装入，而是与之前的数据进行⊕计算。因为接收方顶点会将接收到的信息直接进行⊕计算，作为Δv，所以不妨在发送前就进行⊕计算。）因此最后发送时，worker对每个目的地worker最多发送一条数据。Maiter设置了一个计时器，每隔一段时间就发送所有msg tables中的数据。

Iteration Termination：worker使用progress estimator进行迭代终止管理；master使用global terminator进行全局迭代终止管理。global terminator周期性地向所有worker发送progress request signal信号，并根据每个worker的progress estimator返回的信息判断是否结束迭代。当决定结束迭代时，global terminator广播terminate signal信号，所有worker停止当前操作，将结果数据存入HDFS中。

Fault Tolerance：worker每结束几次迭代，就将当前状态存入HDFS，作为checkpointing。存储的信息除了Local State Table，还有所有msg tables。

Maiter API

Maiter定义了三个模板类API，类中声明了若干纯虚函数作为实现接口。用户将算法转换为DAIC算法时，只需分别继承这三个模板类，并实现其中的纯虚函数。

Partitioner：对输入的图文件（如txt文件）进行处理，生成顶点和边，构造图结构，并对顶点进行分区。
parse_line：读入一行文本，生成顶点和它的所有出边。
partition：根据顶点vid分区。

IterateKernel：将算法描述为DAIC格式，即实现g_{i,j}(x)，⊕，v_j⁰和Δv_j¹四元组。
init：指定v_j⁰和Δv_j¹。
accumulate：实现⊕。
send：实现g_{i,j}(x)。

TermChecker：算法的迭代终止条件。
estimate_prog：返回worker的本地迭代信息。
terminate：master根据每个worker返回的迭代信息，判断是否结束迭代。

下图为PageRank的DAIC算法三个模板类的子类实现。