Source-linked AI summary
Pregelix: Big(ger) Graph Analytics on A Dataflow Engine
Yingyi Bu, Vinayak Borkar, Jianfeng Jia, Michael J. Carey, Tyson Condie
TL;DR
Big-graph analytics need scalable iterative processing, but MapReduce lacks iteration and process-centric systems face memory pressure. Pregelix represents Pregel computations as iterative dataflow and reports strong results across workloads and systems.
Problem
Big-graph workloads need scalable iterative processing, while MapReduce lacks iteration and process-centric systems face high aggregate RAM requirements.
Source: §1 · p. 1, §2.3 · p. 2Method
Results
Pregelix matches Giraph on memory-resident message-intensive workloads, exceeds it by over an order of magnitude on message-sparse workloads, and scales to larger datasets.
Source: §7.7 · p. 11Takeaways & Limitations
The authors conclude that Pregelix combines the Pregel API with data-parallel query evaluation to support out-of-core workloads, scalability, throughput, software simplicity, and runtime flexibility.
Source: §9 · p. 12Pregelix leaves application-specific vertex-deletion semantics, expressed as integrity constraints, to application programmers.
Source: §3 · p. 4
Abstract
from arXiv · showhide
There is a growing need for distributed graph processing systems that are capable of gracefully scaling to very large graph datasets. Unfortunately, this challenge has not been easily met due to the intense memory pressure imposed by process-centric, message passing designs that many graph processing systems follow. Pregelix is a new open source distributed graph processing system that is based on an iterative dataflow design that is better tuned to handle both in-memory and out-of-core workloads. As such, Pregelix offers improved performance characteristics and scaling properties over current open source systems (e.g., we have seen up to 15x speedup compared to Apache Giraph and up to 35x speedup compared to distributed GraphLab), and makes more effective use of available machine resources to support Big(ger) Graph Analytics.
1. INTRODUCTION
Big-graph workloads call for iterative processing, but MapReduce lacks native iteration and process-centric systems can struggle with workloads that exceed memory. Pregelix preserves the Pregel programming abstraction while implementing it as dataflow over Hyracks.
1. INTRODUCTION
- MapReduce lacks iteration and is not ideal for many graph algorithms, motivating specialized systems for large-scale iterative graph processing.Process-centric designs can also struggle when graph workloads are not memory resident.Source: §1 · p. 1
- Pregel lets programmers express graph algorithms through vertex-level user-defined functions, but existing Pregel-like platforms were distinct systems built from scratch.Their process-centric workers load assigned graph partitions into memory, which can limit support for non-memory-resident workloads.Source: §1 · p. 1
- Pregelix models vertex states and messages as tuples, then executes Pregel semantics as an iterative dataflow of relational operators on Hyracks.Message exchange is represented as a join followed by a group-by; alternative physical strategies can serve different workloads.Source: §1 · p. 1
- By leveraging Hyracks operators and access methods, Pregelix reuses components including bulk-data transfer, out-of-core operators, buffer management, indexes, and data shuffle.This avoids building many critical system components from scratch.Source: §1 · p. 1
- Pregelix supports the full Pregel API for in-memory and out-of-core workloads on shared-nothing clusters, with a range of runtime choices.The paper evaluates execution time, scalability, throughput, plan flexibility, and implementation effort, and describes early use cases.Source: §1 · pp. 1–2
2. BACKGROUND AND PROBLEMS
Pregel computations proceed in supersteps over active vertices, while the Google runtime and Giraph use process-centric workers. The section identifies memory pressure, fixed physical strategies, and implementation complexity as shortcomings of these designs.
2.1 Pregel Semantics and Runtime
- Each Pregel superstep runs a user-defined compute function on active vertices, which can process messages, send messages, modify vertex or edge state, change topology, or halt.Vertices reactivate when messaged, and computation terminates when all are inactive and no messages remain in flight; combine and aggregation functions are also supported.Source: §2.1 · p. 2
- The Google Pregel runtime uses a master to coordinate worker supersteps; workers load graph partitions, exchange messages, and checkpoint at user-specified boundaries for fault tolerance.The master also detects termination.Source: §2.1 · p. 2
2.2 Apache Giraph
- Giraph implements Pregel on Hadoop through master and worker tasks, with workers partitioning vertices and exchanging messages over inter-worker channels.Workers report control state, such as active-vertex counts and superstep completion, to the master.Source: §2.2 · p. 2
2.3 Issues and Opportunities
- Process-centric systems impose minimum aggregate RAM requirements, complicating memory-intensive and multi-user workloads; Pregel supports only in-memory computation, while Hama and Giraph have limited or preliminary out-of-core support.The authors attribute reported out-of-memory issues and shortcomings in disk-based processing to ad-hoc implementations.Source: §2.3 · p. 2
- Process-centric designs bind storage, message handling, and redistribution to single implementations, preventing runtime choices tailored to datasets, algorithms, or hardware.Sparse-message algorithms such as single-source shortest paths could benefit from an index of live vertices rather than iterating over all vertices.Source: §2.3 · pp. 2–3
- Implementing a process-centric Pregel runtime requires a complex stack spanning networking, communication, storage, messaging, memory management, and fault tolerance.The section asks whether established data-parallel platforms can simplify implementation.Source: §2.3 · p. 3
3. THE PREGEL LOGICAL PLAN
Pregelix models Pregel execution as a logical query plan over vertex, message, and global-state relations. A full outer join forms compute inputs, while downstream flows update vertices, combine messages, revise global state, and apply graph mutations.
3. THE PREGEL LOGICAL PLAN
- Vertex, Msg, and GS relations represent vertex state, destination-addressed messages, and the program’s global state.Vertices with halt = false are active; GS tracks halting, aggregate state, and the superstep count.Source: §3 · p. 3
- A full outer join pairs messages with destination vertices, preserves messages for missing vertices, and includes active vertices even when they receive no messages.The missing-vertex case creates a vertex with other fields set to NULL.Source: §3 · p. 3
- For each active vertex, compute receives prior messages and outputs updated vertex data, outbound messages, global-state contributions, and graph mutations.The output also supplies a global halt contribution, which is true when the outbound message list is empty and the updated vertex is halted.Source: §3 · p. 3
- The global-state flow combines vertex halt contributions to control another superstep, aggregates contributions for the next global value, and increments the superstep counter.The halt state and aggregate fields depend on compute’s output.Source: §3 · p. 4
- Mutation tuples are grouped by vertex id and resolved before application to Vertex; application-specific vertex deletion semantics remain the programmer’s responsibility.Pregelix leaves those deletion semantics to application programmers as integrity constraints.Source: §3 · p. 4
4. THE RUNTIME PLATFORM
Hyracks executes data-parallel jobs as operator-and-connector DAGs across worker machines. Pregelix relies on its configurable scheduling, storage access methods, operators, connectors, and materialization policies to implement iterative dataflows.
4. THE RUNTIME PLATFORM
- Hyracks runs parallel clones of operator-and-connector DAGs across worker machines, with connectors redistributing data between operators.A master machine orchestrates the worker execution and data exchanges.Source: §4 · p. 4
- User-configurable task constraints let Hyracks schedule operators at specified counts or locations, supporting sticky iterative dataflows.The scheduler solves for a schedule that satisfies the expressed constraints.Source: §4 · p. 4
- B-trees support efficient lookup and scans, while LSM B-trees convert random update I/Os into sequential ones at the cost of potentially slower lookups.LSM B-trees periodically merge an in-memory component with disk components.Source: §4 · p. 4
- Hyracks provides sort-based, HashSort, and preclustered group-by operators, plus merge-based full outer and probe-based left outer index joins.Preclustered group-by assumes input tuples are already clustered by the grouping key.Source: §4 · p. 4
- Connectors repartition, merge ordered streams during repartitioning, or aggregate streams to one receiver; materialization policies are fully pipelined or sender-side materializing pipelined.The sender-side policy writes output to a local temporary file while another thread sends the written data to the receiver.Source: §4 · pp. 4–5
5. THE PREGELIX SYSTEM
Pregelix implements its logical plan on Hyracks through parallel physical-plan clones, distributed storage, and physical-plan alternatives. Its storage choices accommodate different vertex-update and graph-mutation workloads.
5.1 Parallelism
- Physical-plan clones run across Hyracks workers, each handling one data partition and exchanging data through connectors.The described example parallelizes the logical join over two workers using an m-to-n partitioning connector.Source: §5.1 · p. 5
5.2 Data Storage
- Vertex partitions are stored in vid-keyed B-tree or LSM B-tree indexes, with the index choice selectable according to workload.B-trees suit frequent in-place vertex updates; LSM B-trees suit large changes in vertex-data size or frequent graph mutations.Source: §5.2 · p. 5
- Vertex data is partitioned by vid, using hash partitioning by default.The default partitioning function is hash partitioning.Source: §5.2 · p. 5
- Messages are stored in temporary local files on the workers that hold their destination vertices, while HDFS stores global state for worker access and caching.Message partitions are sorted by vid; an access method reads and caches global state when user-defined functions reference it.Source: §5.2 · p. 5
5.3 Physical Query Plans
Pregelix maps logical-plan operations to physical alternatives for combining and delivering messages, revising global state, applying graph mutations, and redistributing data. These choices include workload-dependent strategies and location constraints that avoid unnecessary repartitioning.
5.3.1 Message Combination
- Message combination partitions messages by destination vid for a receiver-side grouping step; four parallel group-by strategies trade stream coordination against receiver-side regrouping.Strategy choice depends on the dataset, graph algorithm, and cluster.Source: §5.3.1 · pp. 5–6
5.3.2 Message Delivery
- Index full outer join scans the vertex index for broadly active workloads, while index left outer join probes live vertices for sparse messages and few active vertices.Users can select the join approach for a job.Source: §5.3.2 · p. 6, Figure 8 · p. 6
5.3.3 Global States and Graph Mutations
- Global halt and aggregate values use two-stage aggregation; vertex additions and removals use index insert-delete, with receiver-side grouping for mutation resolution.Mutation resolution is grouped by vertex id; its resolve function is not guaranteed to be distributive.Source: §5.3.3 · p. 6
5.3.4 Data Redistribution
- Sticky placement keeps Msg and Vertex partitions aligned across supersteps, so required redistributions are outgoing messages and vertex mutations.The join and group-by operators share location constraints; mutations are sent to the partition where they belong.Source: §5.3.4 · p. 6
5.4 Memory Management
Pregelix manages memory for in-memory and out-of-core workloads, checkpoints computation for recovery, and can pipeline compatible jobs to reduce data movement. Pipelining trades off some fault tolerance for improved performance.
5.4 Memory Management
- Buffer caches hold partition pages and spill to disk as needed, while worker memory also serves group-by, network, and file-system caching.The first three memory components are user-tunable; the file-system cache is managed by the operating system.Source: §5.4 · p. 6
5.5 Fault-Tolerance
- Checkpointing Vertex and Msg at user-selected superstep boundaries supports recovery without requiring user programs to handle failures.Recovery reloads Vertex and optional Vid into indexes, and writes checkpointed Msg data to partition-local files.Source: §5.5 · p. 7
5.6 Job Pipelining
- Compatible producer-consumer jobs with the same vertex type can be pipelined without HDFS reads or writes or index bulk-loads, improving performance with reduced fault tolerance.The option was motivated by a genome assembler that chains six graph-cleaning algorithms for many rounds.Source: §5.6 · p. 7
5.7 Pregelix Software Components
Pregelix’s software components collect cluster and job information, manage recoverable failures, schedule workers, generate query plans, and maintain worker runtime context. Its discussion emphasizes out-of-core support, execution flexibility, and reuse of Hyracks libraries without modifying the runtime.
5.7 Pregelix Software Components
- The statistics collector tracks cluster and job metrics, while the failure manager recovers interruption and I/O failures but forwards application exceptions to users.Collected metrics include resource usage and live machines, as well as vertex, edge, and message counts.Source: §5.7 · p. 7
- The scheduler selects workers using collected information, the plan generator creates physical operator DAGs and location constraints, and worker runtime context caches GS and customizes buffer, file, and index management.Pregelix supports the Pregel API in Java, with client-side components and a runtime context on each worker.Source: §5.7 · p. 7
5.8 Discussion
- Out-of-core operators and access methods support disk-based and multi-user workloads while retaining good in-memory performance, and 16 physical configurations tailor execution to algorithms, datasets, and clusters.The configurations combine two vertex-storage options, two message-delivery alternatives, and four message-combination strategies; the implementations reuse Hyracks libraries without runtime modifications.Source: §5.8 · p. 7
6. PREGELIX CASE STUDIES
Pregelix is used for a built-in graph algorithm library, graph-connectivity research, and parallel genome assembly. The cases include algorithm development and testing on a 60-machine cluster, and a genome assembler built on Pregelix.
The Pregelix Built-in Library
- The built-in library includes PageRank, shortest paths, connected components, reachability, triangle counting, maximal cliques, and random-walk graph sampling.The shortest-path implementation uses configurable join, group-by, and connector choices; vertex updates and messages generate dataflow output tuples.Source: §6 · p. 7
Graph Connectivity Problems
- Connectivity building blocks supported algorithms including bi-connected components for undirected graphs and strongly connected components for directed graphs, and were scale-tested on 60 machines with 480 cores and 240 disks.The building blocks included BFS spanning tree, Euler tour, list ranking, and pre/post-ordering.Source: §6 · p. 7
Genome Assembly
- Genomix constructs a De Bruijn graph from billions of genome reads, cleans subgraph patterns, and iteratively merges paths into vertices until one genome sequence remains.Pregelix’s support for adding and removing vertices is heavily used in this case.Source: §6 · p. 7
7. EXPERIMENTS
Across the tested graph workloads, Pregelix matches Giraph on memory-resident message-intensive tasks, scales to larger datasets, and sustains multi-user workloads. The experiments assess execution time, scalability, throughput, physical-plan flexibility, and software simplicity.
7.1 Experimental Setup
- The experiments use a 32-node Linux cluster, Webmap and BTC graphs, and PageRank, SSSP, and connected-components algorithms.For execution-time, scalability, and throughput experiments, Pregelix uses its default plan and the same stated memory settings across experiments.Source: §7.1 · p. 8
7.2 Execution Time
- 3.5× overall and 7× per-iteration speedups over Giraph on SSSP; Giraph fails above a 0.15 dataset-to-aggregated-RAM ratio, while Pregelix scales to out-of-core workloads.For PageRank, Pregelix is up to 2× slower on very small datasets but up to 2× faster on large datasets; for CC, both systems perform similarly fast. GraphLab is up to 5× faster than Pregelix on BTC-Tiny but up to 24× slower on BTC-X-Small.Source: §7.2 · pp. 8–9, Figure 10 · p. 9, Figure 11 · p. 9
7.3 System Scalability
- Pregelix’s SSSP scale-up is closest to ideal because it sends fewer messages; its parallel PageRank speedup is also close to ideal on the tested cluster sizes.For Webmap-X-Small, Giraph, GraphLab, and GraphX show super-linear speedups, consistent with their performance worsening super-linearly as more data is assigned to a worker.Source: §7.3 · p. 9, Figure 12 · p. 10
7.4 Throughput
- Pregelix throughput responds differently to concurrency by workload: it rises for X-Small, Small, and Large Webmap, but drops at the two-job boundary for Medium.For Large, added concurrency raises CPU utilization by about 20% to 30%; for Medium, two concurrent jobs incur about 10GB of I/O each. Giraph, GraphLab, and Hama fail to support concurrent jobs in these experiments.Source: Figure 13 · p. 10, §7.4 · p. 10
7.5 Plan Flexibility
- With Pregelix’s left outer join plan, SSSP is up to 15× faster than Giraph and up to 35× faster than GraphLab per iteration; the best join plan varies by algorithm.The left outer join is faster for message-sparse SSSP, the full outer join wins for message-intensive PageRank, and the two perform similarly for CC. The redistribution-policy tradeoffs also support the need for an optimizer to choose physical plans.Source: §7.5 · pp. 10–11, Figure 15 · p. 11, Figure 14 · p. 11
7.6 Software Simplicity
- Pregelix’s core module contains 8,514 lines of code, compared with 32,197 lines in Giraph-core.Source: §7.6 · p. 11
7.7 Summary
- Pregelix performs comparably to Giraph on memory-resident message-intensive workloads, outperforms it by over an order of magnitude on memory-resident message-sparse workloads, and handles larger datasets and multi-user workloads.The paper also reports over an order-of-magnitude gains over GraphLab, GraphX, and Hama on large datasets for various workloads.Source: §7.7 · p. 11
8. RELATED WORK
Related work spans parallel data-management and dataflow engines, graph-processing systems, and iterative MapReduce extensions. Pregelix is positioned as orthogonal to graph-system contributions and more flexible than approaches limited to MapReduce.
8. RELATED WORK
- The authors report that they no longer have access to a cluster at the scale of their Yahoo! collaborators’ cluster.Source: §8 · p. 11
- Earlier data-management and dataflow systems established partitioned-parallel processing, richer operators and communication patterns, and parallel support for recursion.The paper describes these systems as a foundation for Pregelix.Source: §8 · p. 11
- Graph systems explored partitioning, APIs, asynchronous execution, and distributed storage, while Pregelix targets lower-level out-of-core support, plan flexibility, and software simplicity.The paper characterizes its work as largely orthogonal to those systems and their contributions.Source: §8 · p. 11
- HaLoop and PrIter add iteration-related features to MapReduce, but remain constrained by its model; Pregelix explores more flexible scheduling, storage, operators, and redistribution.HaLoop adds sticky scheduling and caching, while PrIter manages intermediate state and supports user-defined prioritization policies.Source: §8 · pp. 11–12
9. CONCLUSIONS
Pregelix combines the Pregel API with data-parallel query evaluation on Hyracks. The authors report transparent out-of-core support, scalability, throughput, software simplicity, and runtime flexibility, while identifying cost-based plan selection and AsterixDB integration as future work.
9. CONCLUSIONS
- Pregelix combines the Pregel API with database-style data-parallel query evaluation on the Hyracks parallel dataflow engine.The paper presents the system’s design, implementation, early use cases, and evaluation.Source: §9 · p. 12
- Pregelix is reported as the only open-source Pregel-like system that efficiently scales to out-of-core workloads, sustains multi-user workloads, and allows runtime flexibility.Source: §9 · p. 12
- The authors say this architecture could be adopted by parallel data-warehouse vendors to build Big Graph processing infrastructures on existing query engines.They also report several stable open-source releases since 2012 for the Big Data research community.Source: §9 · p. 12
- Future work includes automating physical plan selection with a cost-based optimizer and integrating Pregelix with AsterixDB for richer Big Graph Analytics.Source: §9 · p. 12