Source-linked AI summary
Distributed GraphLab: A Framework for Machine Learning in the Cloud
Yucheng Low, Joseph Gonzalez, Aapo Kyrola, Danny Bickson, Carlos Guestrin, Joseph M. Hellerstein
TL;DR
Existing data-parallel frameworks do not naturally support many important MLDM algorithms. This paper extends GraphLab to distributed execution with consistency-preserving mechanisms and reports large performance gains over Hadoop-based implementations.
Problem
Existing high-level data-parallel frameworks do not naturally or efficiently support many important MLDM algorithms.
Method
The paper extends GraphLab to distributed execution using data versioning, pipelined distributed locking, graph-based data placement, and snapshot-based fault tolerance.
Results
20-60x: GraphLab applications outperform equivalent Hadoop/MapReduce implementations and match carefully constructed MPI implementations.
Takeaways & Limitations
Distributed GraphLab supports asynchronous, dynamic, graph-parallel MLDM computation while preserving strict consistency requirements.
Takeaways & Limitations
The quality of the two-stage partitioning scheme is beyond the paper's scope, with only simple experiments suggesting performance comparable to direct partitioning.
Abstract
from arXiv · showhide
While high-level data parallel frameworks, like MapReduce, simplify the design and implementation of large-scale data processing systems, they do not naturally or efficiently support many important data mining and machine learning algorithms and can lead to inefficient learning systems. To help fill this critical void, we introduced the GraphLab abstraction which naturally expresses asynchronous, dynamic, graph-parallel computation while ensuring data consistency and achieving a high degree of parallel performance in the shared-memory setting. In this paper, we extend the GraphLab framework to the substantially more challenging distributed setting while preserving strong data consistency guarantees. We develop graph based extensions to pipelined locking and data versioning to reduce network congestion and mitigate the effect of network latency. We also introduce fault tolerance to the GraphLab abstraction using the classic Chandy-Lamport snapshot algorithm and demonstrate how it can be easily implemented by exploiting the GraphLab abstraction itself. Finally, we evaluate our distributed implementation of the GraphLab abstraction on a large Amazon EC2 deployment and show 1-2 orders of magnitude performance gains over Hadoop-based implementations.
1. INTRODUCTION
Distributed GraphLab extends the GraphLab abstraction to cloud-scale distributed execution while preserving strict consistency. It addresses distributed systems challenges through specialized execution engines, fault tolerance, and extensive performance evaluation.
- Motivation: Existing high-level frameworks fail to naturally support the asynchronous, dynamic, graph-parallel computation common in MLDM applications.The paper identifies this gap as a source of repeated, time-consuming effort to solve parallel and distributed systems problems for individual models.
- Approach: The distributed GraphLab abstraction provides a formal execution model tailored to distributed MLDM computation.It extends the multi-core abstraction while retaining strict consistency requirements.
- Approach: Data versioning reduces network congestion, while pipelined distributed locking mitigates network latency.These mechanisms are incorporated into the distributed execution model to improve efficiency without abandoning strict consistency.
- Implementation: Fault tolerance is added through two snapshotting schemes, including an adaptation of the Chandy-Lamport snapshot algorithm.The paper demonstrates that fault tolerance can be implemented within the GraphLab abstraction.
- Evaluation: 20-60x: GraphLab applications outperform equivalent Hadoop/MapReduce implementations and match carefully constructed MPI implementations.The evaluation uses an optimized C++ implementation on Amazon EC2, including a 512-processor, 64-node cluster.
- Implementation: The framework adds two graph-based execution engines: Chromatic Engine for static schedules and Locking Engine for dynamically prioritized execution.The Chromatic Engine uses graph coloring, whereas the Locking Engine uses pipelined distributed locking and latency hiding.
2. MLDM ALGORITHM PROPERTIES
The paper motivates GraphLab through four MLDM properties: serializable, graph-structured, asynchronous, and dynamic computation. It contrasts these requirements with frameworks whose execution or consistency models support them incompletely.
- Graph Structured Computation: Graph-structured computation models dependencies among data, which data-parallel abstractions such as MapReduce are generally not designed to express.GraphLab lets vertex computations read and write adjacent vertex and edge data while the runtime manages consistent parallel execution.
- Asynchronous Iterative Computation: Asynchronous computation updates parameters using the most recent available values rather than values from a previous synchronized time step.Synchronous execution can incur penalties because each phase waits for the slowest machine.
- Asynchronous Iterative Computation: Load, network, hardware, multi-tenancy, and kernel-complexity variability make synchronous phases sensitive to stragglers.Power-law graph degree distributions and data-dependent vertex work can skew execution time even under uniform partitioning.
- Asynchronous Iterative Computation: Existing iterative extensions and BSP systems still do not naturally support asynchronous computation.The paper contrasts Spark, Pregel, Piccolo, and BPGL with shared-memory GraphLab's design for asynchronous iterative algorithms.
- Dynamic Computation: Dynamic computation prioritizes work because iterative MLDM algorithms often converge asymmetrically across parameters.In PageRank, most vertices required one update, while about 3% required more than 10 updates.
- Dynamic Computation: GraphLab supports adaptive prioritization and adjacent-data access, whereas Pregel's message-triggered updates restrict the data available to each computation.This distinction makes dynamic PageRank difficult to express in Pregel.
- Serializability: Serializable execution can simplify concurrent MLDM development and is required for statistical correctness in Gibbs sampling.Dynamic ALS is described as unstable under racing, non-serializable execution.
3. DIST. GRAPHLAB ABSTRACTION
The GraphLab abstraction represents machine-learning state and computation on a graph, executes stateless update functions over local scopes, and supports adaptive scheduling. Its execution model combines flexible ordering, serializability, consistency models, and global synchronization to translate sequential semantics into parallel execution.
- Data Graph: The data graph G = (V, E, D) stores mutable user-defined data on vertices and edges while keeping graph structure static during execution.The data can represent model parameters, algorithm state, or statistical data.
- Update Functions: Update functions are stateless procedures that modify a vertex scope and schedule future executions on other vertices.A scope contains the vertex, adjacent vertices, and adjacent edges; the function returns modified scope data and a set of vertices to execute.
- Update Functions: GraphLab supports adaptive pull-style computation by letting update functions read adjacent values independently of whether neighboring vertices scheduled the update.This decouples scheduling from data movement and supports dynamic PageRank computation.
- Execution Model: The execution model repeatedly removes a vertex, applies its update function, and adds returned vertices to the pending set until no vertices remain.Duplicate vertices are ignored, and the completed data graph and global values are returned.
- Execution Model: The runtime may reorder execution to reduce communication or latency and use vertex priorities, while requiring every pending vertex to eventually execute.Multiple processors execute different vertices concurrently, subject to consistency mechanisms that preserve serializability.
- Consistency and Synchronization: Full, edge, and vertex consistency trade stronger access guarantees for less parallelism, greater safe overlap, and maximum parallelism, respectively.Edge consistency permits read-only access to adjacent vertices, while vertex consistency allows all update functions to run in parallel.
- Consistency and Synchronization: Global values are read by update functions and written through sync operations that combine distributed contributions using an associative commutative sum.This supports global statistics such as convergence estimators.
4. DISTRIBUTED GRAPHLAB DESIGN
Distributed GraphLab extends the shared-memory abstraction to distributed in-memory execution while preserving consistency through graph-aware data placement, scheduling, locking, versioning, and snapshot-based fault tolerance. Its design addresses locality, latency, scheduling flexibility, and failures using complementary execution engines and asynchronous coordination.
- Distributed GraphLab extends the shared-memory GraphLab abstraction to distributed in-memory execution, requiring the entire graph and program state to reside in RAM.
- 4.1 The Distributed Data Graph: The atom graph partitions data into many journaled atoms with ghost boundary data, enabling balanced placement and rapid loading across varying cluster sizes.Atoms are stored separately with connectivity metadata, and ghosts cache adjacent remote graph elements.
- 4.1 The Distributed Data Graph: The two-stage partitioning scheme reuses one partition computation for different machine counts without full repartitioning, although its quality study is outside the paper’s scope.
- 4.2.1 Chromatic Engine: The Chromatic Engine uses graph coloring and synchronous color-steps for serializable execution, while supporting different consistency models through different coloring schemes.Communication within color-steps is asynchronous, but a full barrier is required before advancing to the next color.
- 4.2.2 Distributed Locking Engine: The Distributed Locking Engine uses readers-writer locks and pipelined distributed locking to provide scheduling flexibility for dynamically prioritized execution.It addresses limitations of coloring-based execution, including unavailable colorings and insufficient scheduling flexibility.
- 4.3 Fault Tolerance: Distributed GraphLab adds synchronous and asynchronous checkpointing, with the asynchronous method constructing snapshots incrementally without suspending computation.The implementation adapts the Chandy-Lamport snapshot algorithm to the GraphLab data graph and execution model.
- 4.3 Fault Tolerance: A 15-second simulated machine delay added 3 seconds to asynchronous snapshot runtime versus 16 seconds for synchronous snapshots.
5. APPLICATIONS
The evaluation applies GraphLab to three state-of-the-art machine-learning and data-mining applications using large real-world datasets and compares selected implementations with Hadoop and MPI. GraphLab outperforms Hadoop substantially, matches tailored MPI performance, and offers a compact expression of the evaluated algorithms.
- GraphLab was evaluated on collaborative filtering for Netflix recommendations, Video Co-segmentation, and Named Entity Recognition using large real-world problems and datasets.The Chromatic Engine was used for Netflix and NER, while the Locking Engine was used for CoSeg.
- Pregel was not directly compared because it was unavailable and contemporary open-source implementations did not scale to the smaller considered problems.The paper also notes that the considered computations require bidirected edges, increasing graph storage complexity under Pregel’s abstraction.
- 20-60x: GraphLab outperformed equivalent Hadoop implementations on equivalent tasks, while performance was comparable to tailored MPI implementations.
- GraphLab’s performance scaling improved with higher computation-to-communication ratios across the evaluated applications.
- GraphLab expressed the Netflix, NER, and CoSeg algorithms more compactly than MapReduce or MPI.
U у R
Distributed GraphLab represents machine-learning workloads as graph computations and evaluates their scalability, runtime, and communication behavior across several applications. Results show strong gains over Hadoop, while performance depends on graph structure, computation-to-communication ratio, and scheduling.
- Netflix: GraphLab represents ALS as a bipartite graph whose update functions solve least-squares problems for user and movie factors.The latent dimension d controls both approximation accuracy and computational complexity.
- Netflix: 12x at d = 5 and 4.9x at d = 100 quantify Distributed GraphLab overhead at 64 machines as computation per update increases.The overhead was estimated by extrapolating to the theoretically optimal runtime.
- Netflix: 40-60x faster than Hadoop, GraphLab is comparable to the MPI implementation for Netflix at d = 20.The comparison used 4 to 64 machines and a tuned Hadoop implementation with HDFS replication reduced to one.
- Scheduling: Dynamic scheduling reaches low test error in about half the work of a static BSP-style schedule.GraphLab represents this dynamic schedule directly, whereas it is difficult to express using Pregel messaging semantics.
- Video Co-segmentation: 10x speedup with 16x more machines demonstrates scalability for CoSeg on a 10.5 million vertex graph.The locking engine also provides nearly optimal weak scaling because crossing edges increase linearly with machine count.
- Named Entity Recognition: 3x improvement with 16x more machines shows modest NER scaling, with saturation beyond 16 machines at over 100MB per second per machine.The authors attribute this to large vertex data, dense connectivity, and random partitioning, which increase communication overhead.
- Named Entity Recognition: 20-30x speedup over Hadoop was achieved for NER, although MPI outperformed Distributed GraphLab because the task has very little computation relative to data accessed.The NER update requires 5.7x fewer cycles per byte than the hardest evaluated Netflix case.
- Cost evaluation: GraphLab is about two orders of magnitude more cost-effective than Hadoop for Netflix.The price-runtime curve exhibits diminishing returns as reduced runtimes require disproportionately higher cost.
6. RELATED WORK
Related work spans graph databases, MapReduce-based graph processing, and graph-parallel abstractions. The paper positions GraphLab as focused on iterative graph-structured computation rather than graph storage or basic retrieval.
- Scope: The related-work landscape motivates comparing GraphLab with abstractions that differ in graph storage, computation, and scheduling support.The section explicitly reviews classic parallel abstractions, graph databases, and domain-specific languages.
- Graph databases: Graph databases emphasize efficient storage and retrieval of graph-structured data with basic graph computation, whereas GraphLab targets iterative graph computation.Neo4J is cited as an example of a graph database.
- MapReduce systems: MapReduce graph systems include Pegasus, Surfer, and graph-filtering approaches, but Surfer does not support asynchronous or dynamic scheduling.These systems represent alternatives for large-graph processing using Hadoop or preprocessing.
7. CONCLUSION AND FUTURE WORK
Distributed GraphLab extends GraphLab to distributed execution while supporting sparse dependencies, asynchronous computation, dynamic scheduling, serializability, and fault tolerance. Its evaluation shows strong performance against Hadoop and competitiveness with MPI, while future work targets evolving graphs and external storage.
- Distributed GraphLab targets sparse computational dependencies, asynchronous computation, dynamic scheduling, and serializability in large-scale MLDM applications.
- The framework combines a distributed data graph, chromatic and locking engines, and synchronous and asynchronous snapshot-based fault tolerance.The locking engine uses graph-based pipelined locking, while asynchronous snapshots are based on Chandy-Lamport snapshots.
- 20-60x: Distributed GraphLab outperformed Hadoop on three state-of-the-art MLDM algorithms evaluated with real data on up to 512 processors.The evaluation used Amazon EC2 and 64 HPC machines; the system was also competitive with tailored MPI implementations.
- Future work includes support for dynamically evolving graphs and external storage in graph databases.These extensions are intended for continually storing and processing time-evolving data in applications such as social networking and recommender systems.