DDIA：图计算和迭代处理

在图数据建模一节中我们讨论过使用图模型对数据进行建模、使用图查询语言对图中的点边属性进行查询。但第二章相关讨论主要集中在偏 OLTP 方向——对符合要求的小数据集的查询。

在批处理的上下文中，我们可以重新审视图模型——也就是常说的图计算，在全图做一些离线处理和分析。这种需求通常来自推荐系统（比如购物平台的“你可能喜欢”模块）、排名系统中。比如，最出名的图分析算法—— PageRank，最初就是一个对网页权重的排名算法。该算法的基本原理是根据链接到网页的数量和质量，来评估每个网页的受欢迎程度，以最终决定搜索引擎对搜索结果的展示排名。

DAG 和图计算 上一小结提到的 Spark、Flink 和 Tez 等数据流引擎通常以有向无环图（directed acyclic graph，DAG）的形式组织一个计算任务中的算子。但这并不是图计算（graph processing），尽管数据在不同算子间进行流动时，会构成图一样的计算拓扑（SQL 的执行引擎实现也是类似），但这是数据在计算时形成的计算拓扑，而数据集本身的结构仍然是关系型的。但在图计算中，数据本身就具有图结构。比如 PageRank 中的网页间通过链接关系构成的引用图。这又是一个典型的容易引起混淆的、不同领域却具有相似命名的场景。

大部分图计算的算法都是迭代式的，其基本思路是：

1. 每次遍历一条边
2. 和起点进行 join，以传递、连接某些信息
3. 重复 1、2 直到满足某种条件。比如
    1. 遍历完了所有边
    2. 某些指标开始收敛

在第二章的例子中，就是沿着 localion_in 的边来找到所有从属于北美大陆的地点列表。

如果我们想用 Hadoop 生态来进行图计算，使用分布式文件系统存储图数据很容易（比如使用文件来顺序的存点和边），但是使用 MapReduce 来处理这些图数据，就很难表达“不断迭代处理，直到某些条件满足时停止”的语义。因为 MapReduce 只着眼于数据的单次处理，而很难表达这种递归或者迭代的语义。这种迭代风格的算法通常包含以下几步：

1. 执行一轮：全局调度器针对算法的一个步骤调度一个批处理任务。
2. 条件检查：在一次迭代执行完成后，调度器会检查某些条件是否满足，来判断算法是否可以停止。（比如是否还有边需要遍历、结果指标是否收敛等等）。
3. 继续执行：如果结束条件不满足，全局调度器就继续步骤 1 ，调度一轮新的批处理任务。

使用 MapReduce 实现上述过程通常非常低效，因为 MapReduce 在设计时并没有专门面向迭代算法：即使一次迭代只需要增量地读图中的很小一部分数据，MapReduce 也总是无脑读全部输入（本质上是因为 MapReduce 无法针对数据文件中的图结构进行增量调度）。

Pregel 处理模型

BSP（bulk synchronous parallel 模型），作为一种专门为图批处理优化的计算模型，近年来变的越来越流行。Apache Giraph，Spark’s GraphX 和 Flink’s Gelly 都在 API 中实现了该计算模型。由于谷歌的一篇名为 Pregel 的论文将该图计算模型大规模的推广，因此 BSP 模型有时也被称为 Pregel 模型。

在 MapReduce 中，由于 Reducer 需要将具有同样 Key 的数据聚到一块，因此 Mapper 在处理完数据后，会将结果分别“发送” 给对应的 Reducer。Pregel 也有类似的思想——图中的点（vertex）可以“发消息”给其他点。但与 MapReduce 不同的是，由于边这种实体连接的存在，点通常会顺着边发送消息。

在图计算的每一轮迭代中，会对每个点调用回调函数，处理该点收到的消息，这点和 MapReduce 中的 Reducer 很像。但 Reducer 不会保存跨 MapReduce 的状态，但在 Pregel 中，每个点是会保存跨轮次状态（历史处理结果）的，因此可以每次增量式得处理信息。边界情况，如果该节点没有收到任何消息，则无需调用回调函数。

如果你把每个点认为是一个 actor 的话，Pregel 在某种程度上很像我们之前提到的 Actor 模型。但与 Actor 模型不同的是，Pregel 中点的状态和消息是持久化且容错的。此外，Pregel 的通信会以固定的形式执行：执行框架总是将消息从上一轮完全投递到下一轮中。而 Actor 模型则没有该保证。

容错

Pregel 中限定只能通过消息传递（而不是通过主动拉取）来进行通信，因此可以方便的将消息 batch 起来以减少等待。确切的说，所有的等待只存在于相邻的两次迭代之间，因为消息总是从上一轮次的点发出，在通过网络全部发给发给对应节点后，才会开启下一个轮次的计算。这也是 BSP 模型的特点——计算是一轮一轮的，每轮之间存在着一个同步点。

即使在消息传输的过程中，可能会出现丢失、重复和不定时延迟，Pregel 仍然能够保证所有消息在目的节点上严格的被处理一次。和 MapReduce 一样，Pregel 会进行对上层无感的错误恢复，以期简化所有基于 Pregel 的上层算的实现。

容错的方式也很简洁——在每个迭代轮次末尾，将所有顶点的状态做 checkpoint，且持久化到外存。如果某个节点故障，内存中的状态丢失，最简单的恢复方式就是回滚该轮次所有计算，恢复到上一个 checkpoint，然后重启该轮次的所有计算。如果计算是确定性的，且消息也被记录了下来，则代价相对的小的方式是只对故障节点所包含部分的数据进行重新计算（就像之前讨论过的数据流工具，比如 Spark 中的 Partition 容错方式一样）。

并行执行

每个节点并不需要感知其所运行的物理机器；当其想要发消息时，只需要知道下游节点的 VertexID 即可（类似于 MapReduce 中使用 key 进行路由）。如何对图结构（也就是依赖于图顶点的计算）进行划分是框架的职责：

1. 每个顶点运行在哪个机器
2. 每条消息路由到目标顶点

由于模型中的计算只针对单个计算顶点，换句话说，就是每个计算过程都是站在顶点的视角进行“思考”，也即，计算粒度是顶点。这给了框架以任何方式对图结构在不同机器进行划分的自由，理想情况下，将频繁交换信息的节点调度到相同机器上性能最好。但这很难，因为计算是动态的，我们很难事先预知通信的频繁程度，进而依此对顶点进行划分。在实践中，通常使用最简单粗暴的方式，将每个节点随机调度到机器上。

但这样的调度通常会导致大量的跨节点通信开销，甚而，节点间传递的消息规模甚至会比原图数据都大。这种数据传输的额外开销，会非常显著地降低分布式图算法的性能。

故此，如果能将待计算的图数据放进单机内存中，那使用单机图算法效率大概率要比分布式图批处理效率高。即使图结构比单机存大一些，但可以放到单机硬盘上，相较分布式系统，像 GraphChi 这样的单机处理引擎也往往是更好的选择。只有单机存储无法容纳所有数据，再来考虑类似 Pregel 的分布式图计算框架。也因此，最大限度地并行化图计算过程、提升性能的算法，仍然是一个然活跃的研究领域。

参考资料

[1]

DDIA 读书分享会: https://ddia.qtmuniao.com/

DDIA 学习会