Petuum: A New Platform for Distributed Machine Learning on Big Data

Eric P. Xing; Qirong Ho; Wei Dai; Jin Kyu Kim; Jinliang Wei; Seunghak Lee; Xun Zheng; Pengtao Xie; Abhimanu Kumar; Yaoliang Yu

Source-linked AI summary

Petuum: A New Platform for Distributed Machine Learning on Big Data

Eric P. Xing, Qirong Ho, Wei Dai, Jin Kyu Kim, Jinliang Wei, Seunghak Lee, Xun Zheng, Pengtao Xie, Abhimanu Kumar, Yaoliang Yu

arXiv:1312.7651v2Dec 30, 2013stat.MLcs.LGeess.SY

TL;DR

Large-scale ML needs a general way to run diverse optimization-centric programs over Big Data and Big Models despite the limits of conventional and specialized parallel platforms. Petuum addresses this with ML-centric data- and model-parallel execution, bounded-staleness synchronization, and structure-aware scheduling, and reports faster convergence and larger model execution on modest clusters.

Problem
Scaling diverse ML programs to Big Data and Big Models remains difficult because existing platforms offer only partial support for ML-specific error tolerance, fine-grained scheduling, and varied program representations.
Method
Petuum models ML algorithms as iterative-convergent optimization programs and supports data- and model-parallel execution with bounded-staleness synchronization and dynamic parameter scheduling.
Results
Petuum runs a range of ML programs with improved convergence times and large model sizes on relatively modest clusters of 10-100 machines.
Takeaways & Limitations
An ML-centric platform can target fast convergence across diverse data- and model-parallel programs rather than optimizing only operational properties such as fault tolerance or graph consistency.
Takeaways & Limitations
Model-parallel effectiveness requires restricting parallel updates to independent or weakly correlated parameters.

Abstract

from arXiv · show

What is a systematic way to efficiently apply a wide spectrum of advanced ML programs to industrial scale problems, using Big Models (up to 100s of billions of parameters) on Big Data (up to terabytes or petabytes)? Modern parallelization strategies employ fine-grained operations and scheduling beyond the classic bulk-synchronous processing paradigm popularized by MapReduce, or even specialized graph-based execution that relies on graph representations of ML programs. The variety of approaches tends to pull systems and algorithms design in different directions, and it remains difficult to find a universal platform applicable to a wide range of ML programs at scale. We propose a general-purpose framework that systematically addresses data- and model-parallel challenges in large-scale ML, by observing that many ML programs are fundamentally optimization-centric and admit error-tolerant, iterative-convergent algorithmic solutions. This presents unique opportunities for an integrative system design, such as bounded-error network synchronization and dynamic scheduling based on ML program structure. We demonstrate the efficacy of these system designs versus well-known implementations of modern ML algorithms, allowing ML programs to run in much less time and at considerably larger model sizes, even on modestly-sized compute clusters.

1 Introduction

Petuum targets the difficulty of scaling diverse, optimization-centric ML programs by providing a general distributed framework built around iterative convergence and ML-specific execution properties. Its system designs support data- and model-parallel algorithms, with experiments demonstrating faster convergence and large model execution on modest clusters.

Motivation: Big Data and Big Models create space and time bottlenecks that make sequential or batch-style execution inefficient for industrial-scale ML.Examples include 27TB of web text, billions of images, and models with billions to 10^6 topics or more.
Motivation: Migration from specialized academic implementations to corporate clusters or cloud platforms requires difficult distributed programming and low-level resource control.This research-to-production gap limits adoption of advanced probabilistic and optimization programs.
Prior Platforms: Existing platforms provide partial solutions: Hadoop struggles to exploit ML error tolerance, Spark lacks fine-grained scheduling, and graph-centric systems do not fit every ML program.Topic modeling and regression may lack obvious or efficient graph representations.
Petuum’s Principle: Petuum formalizes a broad spectrum of ML algorithms as iterative-convergent programs with explicit objectives over data and model parameters.The framework covers methods including stochastic gradient descent, MCMC, coordinate descent, variational methods, and proximal optimization.
Petuum’s Principle: Petuum exploits a data–model dichotomy: data are conditionally independent and persistent, while model parameters are internally coupled and transient before convergence.This motivates data-parallel and model-parallel execution across diverse ML programs.
System Objectives: Petuum synchronizes parameters with bounded staleness, dynamically schedules according to changing dependencies, and prioritizes parameters that have not converged.These objectives target lower communication and parallelization costs while accelerating convergence.
Evaluation: Petuum implementations scale data- and model-parallel algorithms to large models with improved convergence times on relatively modest clusters of 10-100 machines.The experiments cover topic modeling, matrix factorization, deep learning, Lasso regression, and distance metric learning.

2 Preliminaries: On Data and Model Parallelism

Petuum frames large-scale ML as iterative-convergent optimization over data and model state, exposing distinct data- and model-parallel execution patterns. Data parallelism exploits independent data partitions, whereas model parallelism requires scheduling because parameters may be dependent.

Iterative-Convergent ML: Iterative-convergent ML repeatedly applies an update function to data D and model state A until a stopping criterion is reached.The update improves an explicit objective, and aggregation produces the next model state.
Data Parallelism: Data parallelism partitions D across workers, which independently compute updates on their assigned subsets.Additive updates can be aggregated locally before network transmission, supporting minibatch, asynchronous, and bounded-asynchronous execution.
Model Parallelism: Unlike data samples, model parameters are not generally independent, so effective model-parallel updates must target independent or weakly correlated parameters.Global scheduling can select parameters to reduce parallelization error and avoid non-convergence.
Model Parallelism: Model parallelism partitions model A across workers, with scheduling restricting each update to selected parameters.The scheduling function outputs parameter indices that determine which coordinates each worker updates.
Model Parallelism: Dynamic scheduling can improve performance by prioritizing parameters that are far from convergence while selecting weakly correlated coefficients.In the Lasso example, Petuum combines both criteria to prevent divergence and focus computation.
Platform Requirements: A platform for these programs must synchronize shared state non-blockingly, provide global access to model parameters, and support fine-grained scheduling.The cited discussion identifies missing capabilities in Hadoop, Spark, and GraphLab without code modification.

3 Petuum –

Petuum combines a parameter server, scheduler, and workers to support data- and model-parallel ML programming. Its scheduling interface offers fixed, dependency-aware, and prioritized execution, with pipelining to overlap scheduling and worker computation.

Parameter Server: Petuum provides distributed shared-memory access to global model state through a parameter server interface resembling single-machine programming.The parameter server adopts bounded-asynchronous consistency that preserves data-parallel convergence guarantees.
System Components: Petuum consists of three components: scheduler, workers, and parameter server.The scheduler decides which parameters to update, workers execute updates, and parameter values are delivered through the parameter server.
Scheduling: Fixed schedules dispatch parameters in a predetermined order, including static round-robin execution.This corresponds to the simplest scheduling pattern described for Petuum programs.
Scheduling: Dependency-aware schedules reorder parameter updates using model dependency structure to accelerate model-parallel algorithms such as Lasso regression.Prioritized schedules instead exploit uneven convergence by focusing on selected variables.
Execution: Petuum pipelines compute-intensive scheduling with worker execution so workers continue useful computation while schedules are produced.The scheduler can also perform central aggregation through pull() when required.
Programming Interface: The programming interface includes a central schedule(), parallel push(), and central pull() structure for Petuum programs.The program structure exposes model variables through the parameter-server object.

4 Petuum Parallel Algorithms

Petuum implements both data-parallel and model-parallel ML algorithms through high-level interfaces that expose distributed model updates and scheduling. The examples cover Distance Metric Learning, Lasso, and additional topic-modeling, matrix-factorization, and deep-learning implementations.

4.1 Data-Parallel Distance Metric Learning: Petuum implements data-parallel Distance Metric Learning by relaxing constraints with slack variables and hinge loss, enabling SGD over partitioned sample pairs.Each worker samples local similar and dissimilar pairs and updates the shared factor L.
4.1 Data-Parallel Distance Metric Learning: The DML implementation uses Petuum’s parameter server and bounded-asynchronous consistency to provide stale local parameter copies without per-update network waiting.Only push() requires implementation; schedule() is empty because every worker accesses the full model.
4.2 Model-Parallel Lasso: The Lasso formulation uses coordinate descent with standardized inputs and a convex loss, including squared loss and logistic loss.The paper assumes no intercept and notes that other loss functions can use the same approach.
4.2 Model-Parallel Lasso: Petuum’s model-parallel Lasso scheduler selects nearly independent parameters and prioritizes those farther from convergence to improve optimization.The approach is presented as an improvement over random-subset shotgun coordinate descent.
Additional algorithms: Petuum also implements simultaneous data-and-model-parallel LDA, model-parallel matrix factorization, and data-parallel fully connected and convolutional deep learning.LDA cycles disjoint table and data subsets, matrix factorization assigns factor rows, and deep learning workers update the full model from data shards.

5 Principles and Theory

Petuum frames distributed ML around error tolerance, uneven convergence, and parameter dependencies, then exploits these properties through bounded staleness and dynamic scheduling. The resulting methods provide convergence guarantees while reducing synchronization costs and improving model-parallel execution.

ML programs expose error-tolerant convergence, non-uniform convergence, and dependency structures that can guide distributed execution.
Bounded staleness reduces synchronization overhead while preserving correct outcomes for data-parallel algorithms that tolerate bounded parameter delays.SSP makes workers wait only when a straggler exceeds the staleness threshold.
Under SSP, stochastic gradient descent converges in probability at O(T^-1/2) with an exponential tail bound.Convergence is faster when observed staleness has lower average and variance.
5.2 Dependency structures: Dynamic scheduling selects independent or nearly independent parameters, avoiding the uncontrolled error and possible divergence of naive model-parallel updates.Petuum's scheduler can analyze dependencies at runtime through schedule().
5.2 Dependency structures: SRRP samples a small candidate set, selects features with pairwise correlation below a threshold, and reduces parallelism when enough suitable features are unavailable.The method requires only a small number of correlation evaluations per iteration, unlike preclustering all feature pairs.
5.3 Non-uniform convergence: Prioritizing parameters by distance from convergence accelerates Lasso by decreasing its objective more per iteration.Lasso's sparsity makes many parameters remain at zero, so frequent updates to them can be wasteful.

6 Performance

Across distributed ML workloads, Petuum reports faster convergence, broader model scalability, and near-linear speedups on modest clusters. Its evaluations cover comparisons with established platforms and ports of algorithms lacking prior cluster implementations.

Across ML programs, Petuum runs at least 2-10 times faster than popular implementations.
DML and Lasso: 3.8 times speedup with 4 machines and 1.9 times with 2 machines show that Petuum DML scales with added machines.The same experiments report faster Lasso convergence than Shotgun after scheduler initialization.
Model scalability: Petuum MF is fastest and most memory-efficient against GraphLab and Spark, and uniquely handles Big MF models with rank K ≥1000 on the tested hardware budget.
Model scalability: Petuum LDA converges faster than YahooLDA and supports LDA models more than 10 times larger.The figure compares data-and-model-parallel Petuum with YahooLDA's data-parallel-only implementation.
New cluster implementations: 3.1-times speedup on 4 machines for a Caffe CNN port demonstrates near-linear scaling from a single-machine implementation.The port added approximately 600 lines of Petuum code while retaining the basic data-parallel strategy.
New cluster implementations: Petuum DML converges 3.8 times faster than Xing2002 on 4 machines, supporting efficient cluster implementation of data/model-parallel algorithms.

A Proof of Theorem 2

The SRRP scheduler selects and filters coordinates for parallel coordinate descent, with convergence governed by the interaction between parallelism, dependency structure, and spectral radius. The proof establishes objective decrease under a bounded condition and identifies a parallelization–correctness tradeoff.

Scheduler and dependency control: SRRP randomly selects Q coordinates, retains P after dependency checking, and assigns one retained coordinate to each parallel worker for coordinate descent.The dependency check groups coordinates using the threshold condition |x_i^T x_j| ≤ θ.
Scheduler and dependency control: The scheduler samples dependency-compatible coordinate pairs uniformly, so P varies by iteration and is assumed in practice to equal the number of available workers.The number of eligible pairs can be roughly O(d^2) when θ is close to 1.
Convergence guarantee: When ϵ < 1, the algorithm decreases the objective, which limits the expected number of parallel workers P roughly inverse proportionally to the spectral radius ρ.The objective is F(w) := f(w) + r(w), with w⋆ denoting a global minimizer whose existence is assumed for simplicity.
Convergence guarantee: Larger expected parallelism EP accelerates convergence but increases ϵ, exposing a tradeoff between parallelization and correctness.Smaller variance EP^2 and larger N also improve convergence according to the supplied discussion.
Comparison with prior schedulers: Partitioning zeroes correlations above θ, potentially making SRRP’s spectral radius ρ(A) much smaller than Shotgun’s ρ(X⊤X), while Shotgun cannot control its spectral radius.Compared with the block greedy method, SRRP may also have cheaper per-step work because it picks a random coordinate instead of checking all assigned coordinates.

B Proof of Theorem 3

Theorem 3 compares SRRP with an ideal schedule that selects uncorrelated features. Its analysis shows that SRRP’s parameter trajectory remains close to the ideal trajectory under the stated strong-convexity assumptions.

Theorem statement: Theorem 3 compares SRRP’s parameter trajectory w_RRP(t) with an oracle trajectory w_ideal(t) produced by selecting P random features with zero correlation.The theorem is explicitly framed as SRRP being close to ideal execution.
Proof strategy: SRRP may parallelize fewer than P coordinates when sufficiently independent parameters cannot be found, so its effective parallelism is represented by expected count ˆP.The supplied proof discussion defines ˆP as the expected number of indexes SRRP can actually parallelize in each iteration.
Assumptions: The analysis assumes the objective F(w) = f(w) + r(w) is strongly convex, which can be achieved for certain problems through parameter replication.Replicated Lasso regression is given as an example of this construction.
Proof strategy: The proof bounds the difference between SRRP and ideal trajectories by comparing successive updates, taking expectations over iteration and coordinate-selection randomness, and telescoping the resulting sums.The bound uses data-dependent constants and the expected number of coordinates SRRP can actually parallelize.
Proof strategy: The proof relies on quadratic-objective structure and strong convexity: third- and higher-order derivatives vanish in the Taylor expansion, then the strong-convexity assumption completes the argument.The displayed proof steps invoke Lemma 1, telescoping sums, and finally the strong-convexity constant m.