Source-linked AI summary
Strategies and Principles of Distributed Machine Learning on Big Data
Eric P. Xing, Qirong Ho, Pengtao Xie, Wei Dai
TL;DR
Big Data demands distributed ML systems capable of handling massive datasets and increasingly complex models, while conventional platforms and idealized scalability assumptions leave important system challenges unresolved. The paper distills ML-specific principles for distributing computation, coordinating communication, managing communication, and selecting what to communicate, using them to design efficient and broadly applicable distributed ML systems with convergence and scaling support. Its practical direction combines reusable ML workhorses with an ML-centric cluster operating system, while recognizing that further systems development remains necessary for datacenter deployment.
Problem
Big Data and increasingly complex ML models expose gaps in conventional platforms and scalability approaches that overlook communication costs, heterogeneous progress, and ML-specific computation patterns.
Method
The paper distills ML-rooted statistical and algorithmic principles across distribution, computation–communication bridging, communication management, and communication content.
Results
The paper presents high-performance distributed ML software and general-purpose ML frameworks designed around these principles, including reusable workhorses and an ML Distributed Cluster Operating System.
Takeaways & Limitations
ML systems design can use optimization-centric structure, error tolerance, and algorithmic properties to balance correctness, speed, programmability, and deployability.
Takeaways & Limitations
The envisioned ML Distributed Cluster Operating System still requires containerization, cluster resource management and scheduling, and user-interface development for datacenter deployment.
Abstract
from arXiv · showhide
The rise of Big Data has led to new demands for Machine Learning (ML) systems to learn complex models with millions to billions of parameters, that promise adequate capacity to digest massive datasets and offer powerful predictive analytics thereupon. In order to run ML algorithms at such scales, on a distributed cluster with 10s to 1000s of machines, it is often the case that significant engineering efforts are required --- and one might fairly ask if such engineering truly falls within the domain of ML research or not. Taking the view that Big ML systems can benefit greatly from ML-rooted statistical and algorithmic insights --- and that ML researchers should therefore not shy away from such systems design --- we discuss a series of principles and strategies distilled from our recent efforts on industrial-scale ML solutions. These principles and strategies span a continuum from application, to engineering, and to theoretical research and development of Big ML systems and architectures, with the goal of understanding how to make them efficient, generally-applicable, and supported with convergence and scaling guarantees. They concern four key questions which traditionally receive little attention in ML research: How to distribute an ML program over a cluster? How to bridge ML computation with inter-machine communication? How to perform such communication? What should be communicated between machines? By exposing underlying statistical and algorithmic characteristics unique to ML programs but not typically seen in traditional computer programs, and by dissecting successful cases to reveal how we have harnessed these principles to design and develop both high-performance distributed ML software as well as general-purpose ML frameworks, we present opportunities for ML researchers and practitioners to further shape and grow the area that lies between ML and systems.
1 Introduction
Big Data is pushing ML toward larger datasets and far more complex models, exposing limits in conventional ML research and general-purpose data platforms. The paper argues for ML-specific distributed systems that jointly address scalability, correctness, speed, programmability, and deployment cost.
- Motivation: Big Data and increasingly complex models challenge conventional ML methods designed primarily around model, algorithm, and theory innovations.Applications increasingly involve massive collections and models with billions to trillions of parameters.
- Motivation: Distributed ML scalability depends on both faster convergence and faster per-iteration execution, but idealized assumptions about zero communication cost and uniform machine progress limit industrial applicability.These assumptions are unrealistic on shared research and production clusters.
- System Gap: General-purpose dataflow platforms provide accessible programming and correct execution but underemphasize ML-specific error tolerance and fine-grained computation–communication scheduling, reducing speed.Their abstraction creates a tradeoff among correctness, execution speed, and ease of programming for ML.
- Proposed Direction: A usable ML platform could combine reusable workhorse algorithm implementations with an ML Distributed Cluster Operating System that partitions and executes them across diverse hardware.The proposed combination is intended to reduce both distributed-ML capital costs and human costs.
- Proposed Direction: Future Big Data workloads are expected to shift toward iterative ML operations such as probabilistic inference, constrained optimization, and geometric transformation.These workloads require repeated data passes and high-dimensional mathematical programs, motivating ML-centric architecture design.
2 Background: Iterative-Convergent ML Algorithms
ML programs can be expressed as iterative optimization procedures that repeatedly compute updates from data and model state, then aggregate them into new parameters. Their distinctive data, model, objective, and dependency structures shape how they can be parallelized, including simultaneous data-and-model partitioning for LDA.
- Optimization-centric programs: Most ML programs optimize an objective combining a data-fitting loss with a structure-inducing function that encodes constraints or domain knowledge.The data may be labeled or unlabeled, while the model and objective can be massive or structurally complex.
- Iterative-convergent form: Each iteration computes an update from the previous model state and data, then applies an aggregation function to produce the next model parameters.The update function produces intermediate results intended to increase the objective, while aggregation forms A(t).
- Concrete examples: Lasso uses block coordinate descent with soft-thresholding, whereas LDA can use collapsed Gibbs sampling over token topics, document-topic vectors, and word-topic vectors.These are distinct algorithmic techniques applied to different ML objective structures.
- Parallel execution: Data parallelism partitions samples, while model parallelism partitions parameters and schedules workers onto subsets that are mutually independent or weakly correlated.Data parallelism exploits outer summations over i.i.d. samples; model parallelism requires parameter selection to avoid conflicting updates.
- Parallel execution: LDA requires simultaneous data-and-model parallelism because token updates can touch both word-topic columns and document-topic rows, enabling near-perfect speedup with P machines when partitioned block-wise.The required partitions separate vocabulary columns and document rows so workers avoid operating on the same blocks.
3 Principles of ML System Design
Distributed ML system design organizes four decisions: how to distribute computation, bridge computation with communication, communicate between machines, and choose what to communicate. The paper presents scheduling and bounded-asynchrony principles that preserve convergence while improving execution speed.
- How to distribute: Ideal Lasso schedules partition weakly correlated parameter pairs into subsets that workers update in parallel while preserving sequential dependencies within each subset.Computing all O(m^2) pair correlations is intractable for high-dimensional problems, motivating quickly computable near-ideal scheduling.
- How to distribute: SAP combines parameter prioritization, dependency-aware scheduling, and load balancing into one programmable abstraction for distributed ML.The three functions are intertwined: prioritized parameters affect dependency checks, while independent subsets affect load balancing.
- How to distribute: SAP is near-optimal: its parameter estimate approaches the ideal oracle estimate at O(t^-2), while slow-worker agnosticism preserves convergence and improves progress per iteration.Theoretical results attribute faster convergence to additional updates that reduce model variance.
- How to distribute: Empirically, SAP systems achieve order-of-magnitude speedups, with Strads converging faster than competing implementations on Lasso, Matrix Factorization, and LDA.Strads realizes SAP schedules across all three programs.
- How to bridge computation and communication: BSP waits at iteration barriers, whereas asynchronous execution avoids waiting but can let workers drift far enough apart to cause unrecoverable errors.BSP can also lose efficiency because stragglers and communication barriers delay every worker.
- How to bridge computation and communication: SSP bounds worker divergence to s iterations, combining asynchronous communication with forced pauses that support convergence guarantees.SSP includes BSP as the special case s = 0 and provides guarantees for both data- and model-parallel execution.
4 Petuum: a Realization of the ML System Design Principles
Petuum realizes the paper’s ML system design principles by combining ML-aware execution with reusable distributed infrastructure. It targets both faster iterative learning and broader usability across large clusters and ML algorithms.
- Petuum aims to unify high-performance specialized systems and easier but slower general-purpose platforms through an ML-centric distributed cluster operating system.
- Petuum exploits error tolerance, dependency structures, non-uniform convergence, and compact updates to improve ML convergence rate and per-iteration time.
- Petuum achieves close to ideal P-fold speedup with P machines while supporting clusters ranging from 10s to 1000s of machines.
- Petuum combines Bösen, a bounded-asynchronous distributed key-value store, with Strads to support data-parallel ML programming.
- Its ML library provides over 10 ready-to-run algorithms, spanning logistic regression, k-means, random forest, topic models, deep learning, metric learning, and sparse coding.
- The envisioned platform still requires containerization, cluster resource management, scheduling, and user interfaces to reduce the operational cost of massive-scale ML deployment.