Experimental Analysis of Distributed Graph Systems 分布式图（计算）系统的（效率）实验分析 滑铁卢大学 2019/03/18 赵子豪

背景 Introduction about: Graph process model and system Workload Dataset Experiment design Result & Analysis： Blogel: The Overall Winner GraphX is not efficient when large number of iterations are required

Vertex-Centric BSP Giraph：map-only，edge-cut，compute 基于节点的整体同步并行计算模型 Each vertex computes its new state based on its current state and the messages it receives from its neighbors. Each vertex then sends its new state to its neighbors using message passing. Giraph：map-only，edge-cut，compute GraphLab/PowerGraph: C++,MPI,GAS，vertex-cut (Asynchronous) Blogel-V： C++，MPI，Vertex-cut 每个节点基于现有状态和从邻居收到的信息更新状态，然后将新的状态传播给邻居节点。 同步版本在每一个迭代步骤结束的时候在机器间同步信息 Giraph：是一个基于Hadoop的map-only的因公，把所有数据加载到内存中。使用edge_cut随机划分 GraphLab：Gather，Apply，Scatter，使用节点划对大度节点支持比较好。自动使用所有核和内存。有异步版本，在一轮迭代中，可以获取其他节点的最新信息。

Block-Centric BSP Blogel-B： Also known as graph-centric. Partition the graph into blocks of vertices. Run a serial algorithm within a block while synchronizing blocks on separate machines using BSP. To: Reduce the number of iterations. Blogel-B： Compute function：a serial algorithm runs within the block. Graph Voronoi Diagram：multiple connected components. 划分成block 在一个block里边运行一系列算法 Block数肯定比节点数少，所以网络通信开销就小 GVD划分方法，多个连通子图

MapReduce/Optimized HaLoop： Hadoop Not suitable for iterative workload, I/O 提供Map和Reduce两个接口，极大便利并行处理。Hadoop是最有名的开源实现。因为I/O开销大，还有shuffling的耗时大，Hadoop不适合用迭代轮次多的图算法。但是可以用于内存不够的场景中。 在map和reduce步骤之间缓存可以复用的数据，避免对无关数据的重复扫描以及在机器之间的shuffling操作。 Hadloop:主要目的是减少数据shuffling和reduce工作，在迭代负载上做出如下改进以提高性能：1. 适合迭代变成的模型：比如在主节点增加循环控制2. 改进任务调度器，减少网络通信。3. 从节点上缓存一些loop-invariant的数据并建立索引。4. 最后一轮迭代的结果缓存在本地，用于后续比较，无需去HDFS里边取回。 在内存中划分数据集。RDD。使用节点划分，每一轮迭代包含多个Spark jobs。 HaLoop： Caching reusable data between map and reduce steps Reduce data shuffling and reduce network usage after the first iteration Spark/GraphX RDD Vertex-cut

Relational Vertica： Optimize for Repeated joining. Create a new table. Relational database Distributed. Not HDFS Vertica，这种系统使用关系型数据库存储数据（数据库实现了分布式存储）。重复的join操作效率低。 vertica做出如下改进：1. 在vertex table上更新多个值开销比创建新表更大。超过阈值就创建新表，而非修改值。2. 在一些遍历类的任务（SSSP）中每轮迭代只用一部分数据，将活跃点保存在一个临时的table中。 Vertica： Optimize for Repeated joining. Create a new table.

Stream Systems Flink Gelly： Stream/Batch Flink TensorFlow 定义操作，形成拓扑图，数据在拓扑图中被处理、传递并形成数据流。本文分析了Flink Gelly。两种approach：stream和batch。stream可以从数据源以流的形式加载数据，一边加载，一边处理，这样不容易区分准备数据和实际计算的时间。为了保证和其他工具一致，文中使用的是batch，这样可以把准备数据的时间和计算时间分开。 Flink Gelly： Stream/Batch

Workloads PageRank WCC SSSP：Random Start K-hop：Random Start SSSP和K-hop都使用随机起点

Dataset

Experiment Design 数据加载时间 结果保存时间 执行时间 总体响应时间 Amazon EC2 AWS r3.xlarge machine. 4 cores, 30.5G memory, Xeon E5-2679 v2, SSD disk. Single Master，16、32、64、128machines.

Result

Result

Result

Result

Result

Result Blogel: The Overall Winner Blogel-Vertex: Best end-to-end performance. Only one to finish the SSSP/WCC across all cluster size over WRN dataset. Only one to finish computations over ClueWeb in the 128-machine Clusters. Blogel-Block: Shortest execution time for WCC, SSSP and K-hop Blogel-Vertex: 端到端的性能最好 使用WRN数据集，唯一在所有节点上都完成了SSSP和WCC测试的系统。 唯一可以在128台机器的集群上完成ClueWeb计算的。 因为没有昂贵的“infrasturcture”（such as Hadoop or Spark）。使用高效的C++库，优化利用所以CPU核，有小的内存footprint。 BB:在一些可达性的查询上（SSSP,WCC,K-hop等）有最好的表现，因为：这些查询从Voronoi划分上收益,基于Block的计算最小化了网络通信的开销，而且降低了网络通信 BB运行PageRank的表现不好，因为在block上跑是为了给全局计算提供一个好的初始值，但是这种算法没有产生好的初始值，导致全集性能降低，导致需要更多次迭代。

Result GraphX: The slowest one. GRAPHX/SPARK比所有其他系统都慢。因为spark overhead，data shuffle,长的RDD血统，和checkpoint。以前的研究表明，GRAPHX高效，因为它使用了一个特殊的Spark原型版本，in-Memory的shuffle。此功能在最新版本不可用GraphX在所有规模的集群上基于WRN做WCC都失败了，因为超过了内存大小，有的是超时。结果证明，spark保持RDD血统的容错机制导致了内存错误。当迭代次数增加时，这些lineage就变成了内存占用的大户，导致了潜在的内存不足错误。最近引入的Graphframes[11]对于graphx应该是一个更有效的选择。我们研究了Graphframes的实现，发现它的许多算法可以将输入的图转换为GraphX格式，然后运行GraphX算法。我们还发现，大多数算法对迭代次数有一个默认的最大限制，以减少RDD中长lineage的潜在开销。例如，sssp的限制为10，否则它将开始进入一个checkpoint。此外，WCC的实现要求每两轮迭代就进行一次检查点。检查点可以避免lineage变得太长，但是导致了昂贵的I/O开销，着就会导致超时错误。GraphFrames提供了一个hash-min算法去计算WCC，这种方法迭代次数少，经测试与Blogel性能差不多。还注意到，spark不能在节点之间做负载均衡，一些机器分的partition很多，在同步计算模型，最慢的节点拖慢了整体进程。在实际情况中，花分数不应该超过输入文件中的block数，因为这会导致spark多次读同一个data block。在另一方面，划分数也不应该比集群中的核数少，这导致CPU使用率低。 J. E. Gonzalez, R. S. Xin, A. Dave, D. Crankshaw, M. J. Franklin, and I. Stoica. Graphx: Graph processing in a distributed dataflow framework. In Proc. 11th USENIX Symp. on Operating System Design and Implementation, pages 599–613, 2014. D. A. Bader, G. Cong, and J. Feo. On the architectural requirements for efficient execution of graph algorithms. In Proc. of Parallel Processing, pages 547–556, 2005

Result System Overhead： Giraph（Hadoop） and Graph（Spark） Blogel and GraphLab（MPI） Flink Syncchronous vs. Asychronous： GraphLab Distributed lock contention. 对Giraph和Spark来说，系统overhead比较大，因为Hadoop和spark启动任务，结束任务的开销都很大。使用mpi通信的，部署的时间节省很多。Flink没有很好的内存回收机制，导致内存溢出错误。 GraphLab是唯一一个提供了异步计算模式的系统。但是在每台worker节点上开了数以千计的新进程处理节点，导致分布式加锁争用问题。因此PageRank的异步计算比同步计算更慢。还发现异步计算只对特殊规模的集群适用。具体规模需要尝试。

THANK YOU