Strategies & Principles for Distributed
Machine Learning

Eric Xing
epxing@cs.cmu.edu
School of Computer Science

Carnegie Mellon University
Acknowledgement:
Wei Dai, Qirong Ho, Jin Kyu Kim, Abhimanu Kumar, Seunghak Lee, Jinliang Wei, Pengtao Xie, Yaoliang Yu, Hao Zhang, Xun Zheng
James Cipar, Henggang Cui,and, Phil Gibbons, Greg Ganger, Garth Gibson

1

Machine Learning:-- a view from outside2

Inside ML …

• GraphicalModels• Nonparametric• Regularized• Spectral/Matrix• Deep Learning
Bayesian Models Bayesian Methods• Large-Margin
Methods • Sparse Structured• Sparse Coding
I/O Regression

Hardware and infrastructure• Network switches • Network attached storage • Server machines • GPUs
• Infiniband• Flash storage• Desktops/Laptops• NUMA machines• Cloud compute • Virtual Machines(e.g. Amazon EC2)

3

An ML Program~ + ⌦(✓)~
arg max ⌘ L({xi , yi}N
;
✓)
i=1
~
✓

Model

Data

Parameter

Solved by an iterative convergent algorithm

for	
  (t	
  =	
  1	
  to	
  T)	
  {	
  
	
  	
  doThings()	
  
t+1
tt
t+1
~
~
~
~
	
  	
  
✓	
  ✓ ==g(✓✓ +,	  	
  doOtherThings()	
  

	
  	
  

~
~
✓(D)
✓(D))
ff

}	
  
This computation needs to be parallelized!

4

Massive Data
32 million
pages
1B+ USERS
30+ PETABYTES

100+ hours video
uploaded every minute

645 million users
500 million tweets / day5

Growing Model Complexity
Google Brain
Deep Learningfor images:
1~10 Billion
model parameters

Multi-task Regression
for simplest wholegenome analysis:
100 million ~ 1 Billion
modelparameters

Topic Models
for news article
analysis:
Up to 1 Trillion
modelparameters

Collaborative filtering
for Video recommendation:
1~10 Billion
modelparameters
6Challenge #1– Massive Data Scale

Δf θ(

for	
  (t	
  =	
  1	
  to	
  T)	
  {	
  
	
  	
  doThings()	
  	  	  parallelUpdate(x,θ)	
  	  	
  doOtherThings()	
  
}	
  

D)
Source: Cisco Global Cloud
Index

Source: The Connectivist

Familiar problem: data from 50B devices, data
centers won’t fit into memory of single machine

7

Challenge #2– Gigantic Model Size

for	
  (t	
  =	
  1	
  to	
  T)	
  {	
  
	
  	
  doThings()	
  	  	  parallelUpdate(x,θ)	
  	  	
  doOtherThings()	
  
}	
  

Source: University of
Bonn

Δf

θ(D)

Big Data needs Big Models to extract understanding
But ML models with >1 trillion params also won’t fit!

8

Issues with Hadoop and
I-C ML Algorithms
Iteration 1

Iteration 2

Image source: dzone.com

HDFS BottleneckNaïve MapReduce not best for ML●●

Hadoop can execute iterative-convergent, data-parallel ML...
o map() to distribute data samples i, compute update Δ(Di)o reduce() to combine updates Δ(Di)o Iterative ML algo = repeat map()+reduce() again and again
But reduce() writes to HDFS before starting next iteration’s map() - very slow iterations!

9Spark: Faster MapR on Data-Parallel● Spark’s solution: Resilient Distributed Datasets (RDDs)ooo

Input data → load as RDD → apply transforms → output result
RDD transforms strict superset of MapR
RDDs cached in memory, avoid disk I/O● Spark ML library supports data-parallel ML algos, like HadoopooSpark and Hadoop: comparable first iter timings…
But Spark’s later iters are much faster

Source: ebaytechblog.com

10Parallelization Strategies

for	
  (t	
  =	
  1	
  to	
  T)	
  {	
  
	
  	
  doThings()	
  	  	  parallelUpdate(x,θ)	
  	  	
  doOtherThings()	
  
}	
  

Usually, we worry …

1

2



A parallel program⌘?

but assuming an ideal system, e.g.,

1

Sync

1

 uniform local progress

2

Unequalperformance zero-cost sync,
 zero-cost fault recovery

2

Low bandwidth,High delay

Compute vs Network
LDA 32 machines (256 cores)

Seconds

A sequential program

8000
7000
6000
5000
4000
3000
2000
1000
0

Network waiting time
Compute time

0

8

16

24

 …

11

32

40

48

Intrinsic Properties of ML Programs[Xing et al., 2015]

 ML is optimization-centric, and admits an iterative convergent

algorithmic solution rather than a one-step closed form solution
 Error tolerance: often robust against limited

errors in intermediate calculations
 Dynamic structural dependency:changing correlations between model parameters
critical to efficient parallelization
 Non-uniform convergence: parameters

can converge in very different number of steps

 Whereas traditional programs are transaction-centric, thus only

guaranteed by atomic correctness at every step How do existing Big Data platforms fit the above?

12

4 Principles of ML System Design
How to execute distributed-parallel ML programs?
ML program equations tell us “What to Compute”. But…
1. How to Distribute?
2. How to Bridge Computation and Communication?
3. How to Communicate?
4. What to Communicate?

13Principles ofML system Design[Xing et al., to appear 2016]

1. How to Distribute:
Scheduling and Balancing workloads

14

Example: Model Distribution
Lasso via coordinate descent:min ky

N

y =

X

2
k2 +

X
j

X
M

A huge number of parameters(e.g.) M > 100 million

| j|
Model

M

b0

b2

b4

b5

b6

b7

b10

b11

b1

b3

G0
b8

b9

G1

• How to correctly divide
computational workload
across workers?
• What is the best order toupdate parameters?

15

Model Dependencies
 Concurrent updates ofSequential updates

1

may induce errors

Concurrent updates

1

2
Sync

1

2

2

Need to check x1Tx2
before updating
parameters

Decreases iteration progress(t)
1

S(xT1 y(t 1)
Tx1 x2 2,)16

Avoid Dependency Errors via
Structure-Aware Parallelization (SAP)
key-valuestorescheduler[Lee et al., 2014] [Kim et al, 2016]

data
partition

model
partition

data
partition
key-value
store

model
partition

data
partitionscheduler

model
partition

workerworkerworkerdata
partition

data
partition

data
partition

model
partitionscheduler model
partition

workerworkerworkerdata
partition

data
partition

data
partition

model
partition

model
partition
workerkey-valuestore

qmodelSmart model-parallel
execution:
modelq Structure-aware
schedulingpartition
partitionq Variable prioritization
workerq Load-balancing

workerq Simple programming:q Schedule()q Push()q Pull()17

S. Lee, J.-K. Kim, X. Zheng, Q. Ho, G. Gibson, and E. P. Xing. On Model Parallelization and
Scheduling Strategies for Distributed Machine Learning. NIPS 2014.

A Structure-aware Dynamic Scheduler(Strads) [Lee et al., 2014] [Kim et al, 2016]
Strads System• Priority Scheduling(1) Partition Data + Model into Tasks(2) Schedule & Prioritize Tasks onto Workers

SAP{ j} ⇠(3) Balance Task Load on each Worker

Load-balanced Tasks

Worker 1

⇣

(t 1)
j

⌘2

+⌘

• Block schedulingSync.
barrier

Worker 2
Worker 3[Kumar, Beutel, Ho and Xing, Fugue:
Slow-worker agnostic distributed
learning, AISTATS 2014]

Worker 4
Round 1Round 2Round 3Round 418

SAP Scheduling: Faster, Better
Convergence across algorithms
 SAP on Strads achieves better speed and objective

80 ranks
9 machines

100M features
9 machines

0.15

1.5
1

0.1

0.5

0.05
0

500
Seconds

Lasso

1000

−2.5

STRADS
GraphLab

2
RMSE

0.2

Objective

2.5

STRADS
Lasso−RR

Log−Likelihood

0.25

0

50

100
Seconds

MF

2.5M vocab, 5K topics
32 machines

9

150

x 10

−3

STRADSYahooLDA

−3.5
0

1

2

3
4
Seconds

LDA

5
4
x 10

19

SAP gives Near-Ideal
Convergence Speed[Xing et al., 2015]

 Goal: solve sparse regression problem
 Via coordinate descent over “SAP blocks” X(1),	
  X(2),	
  …,	
  X(B)	
   X(b) are data columns (features) in block (b)	   P parallel workers, M-dimensional data ρ = Spectral	
  Radius[BlockDiag[(X(1))TX(1),	
  …,	
  (X(t))TX(t)]]; this block-diagonal

matrix quantifies max level of correlation within all SAP blocks X(1),	
  X(2),	
  …,	
  X(t)	
   SAP converges according to
Gap between current
parameter estimate and optimum

SAP explicitly minimizes ρ, ensuring
as close to 1/P convergence as possible

where t is # of iterations
 Take-away: SAP minimizes ρ by searching for feature subsets X(1),	
  X(2),	
  …,	
  X(B) w/o cross-correlation => as close to P-fold speedup as possible	
  20

Correctly Measuring Parallel
Performance[blinded, to appear]

YahooLDA progress per iteration
Decreasing
progress periteration
SAP-LDA, m=25
SAP-LDA, m=50
SAP-LDA, m=100

80GB data, 2M words,
1K topics, 100 machines
YahooLDA data throughput
25 machines

39.7 M/s (1x)50 machines

78 M/s (1.96x)

100 machines

151 M/s (3.8x)

Iterations
 YahooLDA attains near-ideal throughput (1→3.8x)…
 … but progress per iteration gets worse with more machines
 YahooLDA only <2x speedup from 25 →100 machines

6.7x slower compared to SAP-LDA

21

Correctly Measuring Parallel
Performance[blinded, to appear]

SAP-LDA progress per iteration
Overlapping
curves – perfectprogress periteration
SAP-LDA, m=25
SAP-LDA, m=50
SAP-LDA, m=100

80GB data, 2M words,
1K topics, 100 machines
SAP-LDA data throughput
25 machines

58.3 M/s (1x)50 machines

114 M/s (1.96x)

100 machines

204 M/s (3.5x)

Iterations


Ideal rate: progress per iter preserved from 25 → 100 machines Thanks to dependency checkingNear-ideal throughput: data rate 1x → 3.5x from 25→100 machines Thanks to load balancing

Convergence Speed = rate x throughput
 Therefore near-ideal 3.5x speedup from 25→100 machines22Principles ofML system Design[Xing et al., to appear 2016]

2. How to Bridge Computation and Communication:
Bridging Models and Bounded Asynchrony

23

The Bulk Synchronous Parallel
Bridging Model[Valiant & McColl]

Thread 1

1

Thread 2

1

Thread 3

1

Thread 4

1

split

Input
Data

Input
Data
Input
Data

2

3
2

3

2

3

2

Update local
copy of ALL
paramsUpdate local
copy of ALL
params3

aggregate
UpdateALLparams Perform barrier in order to communicate parameters Mimics sequential computation – “serializable” property
 Enjoys same theoretical guarantees as sequential execution

24

The Bulk Synchronous Parallel
Bridging Model[Valiant & McColl]

Thread 1

1

Thread 2

1

Thread 3

1

Thread 4

1

2

3
2

3

2
2

3
3

The success of the von Neumann model of sequential computation
is attributable to the fact it is an efficient bridge between softwareand hardware… an analogous bridge is required for parallel
computation if that is to become as widely used – Leslie G. Valiant
 Numerous implementations since 90s (list by Bill McColl): Oxford BSP Toolset (‘98), Paderborn University BSP Library (‘01), Bulk Synchronous Parallel

ML (‘03), BSPonMPI (’06), ScientificPython (’07), Apache Hama (’08), Apache Pregel (‘09),
MulticoreBSP (’11), BSPedupack (‘11), Apache Giraph (’11), GoldenOrb (‘11), Stanford GPS
Project (‘11) …

25

But There Is No Ideal Distributed
System!
 Two distributed challenges:
 Networks are slow
 “Identical” machines rarely perform equally

Result: BSP barriers can be slow
Unequalperformance

Seconds

Low bandwidth,High delay

Compute vs Network
LDA 32 machines (256 cores)
8000
7000
6000
5000
4000
3000
2000
1000
0

Network waiting time
Compute time

0

8

16

24

32

26

40

Is there a better way to interleavecomputation and communication?
 Safe/slow (BSP) vs. Fast/risky (Async)?
 Challenge 1: Need “Partial” synchronicity
 Spread network comms evenly (don’t sync unless needed) Threads usually shouldn’t wait – but mustn’t drift too far apart!

 Challenge 2: Need straggler tolerance
 Slow threads must somehow catch up

BSP
1

Thread 1
Thread 2

1

Thread 3

1

Thread 4

1

2

Async
Thread 1

3
2

3

2

3
3

???

1

Thread 2

1

2

Thread 3

1

2

Thread 4

2

3

4

3
3

4
4

5
5

5

6
6

6

Is persistent memory really necessary for ML?

27

A Stale Synchronous Parallel
Force stop worker 1 until
Bridging Model
worker 2 catches up[Ho et al., 2013]

Staleness Threshold s = 3	

BSP
Worker 1

1

Thread 1

2

3

2

Thread 2

Thread 3

1

3

2

3

3

Thread 4

Worker 2

Async

Worker 3
Thread 1

Thread 2

Worker 4

1

2

Thread 3

3

5

6

4

Thread 4

0

1

2

3

4

5

6

7

8

9

Iteration

Stale Synchronous Parallel (SSP)•

Fastest/slowest workers not allowed to drift >s iterations apart

Consequence••

Fast like async, yet correct like BSP
Why? Workers’ local view of model parameters “not too stale” (≤s iterations old)28

Data-Parallel
Proximal Gradient under SSP
 Model (e.g. SVM, Lasso …):
data D, model a	

sub-update Algorithm:
 Updateproximal step wrt gstale sub-updates Δ() received
by worker p at iteration t

 sub-update

gradient step wrt f

 Data parallel: Data D too large to fit in a single worker, divide among P workers
split

Input
Data

Input
Data
Input
Data

Update local copyof ALL params

Update local copyof ALL params

aggregate
UpdateALLparams29

SSP Data-Parallel
Async Speed, BSP Guarantee
LDA

Lasso

Matrix Fact.

 Massive Data Parallelism
 Effective across different algorithms

30

SSP Data Parallel Convergence Theorem[Ho et al., 2013, Dai et al., 2015]

Let observed staleness be
Let staleness mean, variance be,

Theorem: Given L-Lipschitz objective ft and step size ht,

where

	

Explanation: the distance between true optima andcurrent estimate decreases exponentially with more
SSP iterations. Lower staleness mean, variance,improve the convergence rate.
31Model-Parallel
Proximal Gradient under SSP
 Model (e.g. SVM, Lasso …):
data D, model a	 Model parallel Model dimension d too large to fit in a single worker
 Divide model among P workers

 Algorithm:
on worker pworkers can skip updates

staleness

gradient step wrt f

proximal step wrt g worker p keeps local copy of the full model (can be avoided for linear models)32

SSP Model-Parallel
Async Speed, BSP Guarantee
Lasso: 1M samples, 100M features, 100 machines

Curves overlap – nocompromise to quality

2x speedup

 Massive Model Parallelism
 Effective across different algorithms

33

SSP Model Parallel Convergence Theorem[Zhou et al., 2016]

Theorem: Given that the SSP delay is bounded, withappropriate step size and under mild technical conditions,
then
Finite length
1
1
X
t=0

kx(t + 1)

x(t)k < 1X
t=0

kxi (t + 1)

xi (t)k < 1

In particular, the global and local sequences converge to thesame critical point, with rate O(t-1):

Explanation: Finite length guarantees that the algorithm
stops (the updates must eventually go to zero).
Furthermore, the algorithm converges at rate O(t-1) to the
optimal value; same as BSP model parallel.

34

Principles ofML system Design[Xing et al., to appear 2016]

3. How to Communicate:Managed Communication and Topologies

35

Managed Communication[Wei et al., 2015]

 SSP only Communicates only at iteration boundary Ensures bounded staleness consistencycomputecomputecompute SSP + Managed Communication
 Continuous communication/synchronization
 Update prioritization
 Same consistency guarantees as SSP
computecomputecompute36

SSP
Bounded StalenessManaged Communication,
SSP + Managed200Mbps
Comms
Already enjoying
SSP speedup

1e+08

training loss

Lower
is better

1e+09

MatrixFact:
Managed Communication Speedup

Further 1.8x speedup
StoppingoverCriteria
multiplier
SSP

1e+07

1.8x

•
•

0

200

400

600

800

1000 1200 1400 1600

time (seconds)

Matrix Factorization, Netflix data, rank = 400
8 machines * 16 cores, 1GbE ethernet
37

LDA:
Managed Communication Speedup
time to convergence (seconds)

3000

Already enjoying
SSP speedup

SSP
BoundedStaleness
SSP +Randomized
MC (no prio.)
SSPRelativeMagnitude+ Managed Comms

2500
2000

3x additional speed up fromcomms management

1500

25% additional speedupfrom comms prioritization

1000
500
0

NoMgt

320Mbps

640Mbpsexecution mode••

Latent Dirichlet Allocation, NYTimes, # topics = 1000,
16 machines * 16 cores, 1GbE ethernet
38

Topology: Master-Slave
Model partition

Model partition

server 1

server 2

worker 1

worker 2

ML AppClient lib

Data partition

ML AppClient lib

Data partition

 Used with centralized storage paradigm
 Topology = bipartite graph: Servers (masters) to Workers (slaves) Disadvantage: need to code/manage clients and servers separately
 Advantage: bipartite topology far smaller than full N2 P2P connections

39

Topology: Peer-to-Peer (P2P)
worker 1

worker 2

ML AppModel copy

ML AppModel copy

ML AppModel copy

ML AppModel copy

worker 3

worker 4

 Used with decentralized storage paradigm
 Workers update local parameter view by broadcasting/receiving
 Disadvantage: expensive unless updates ΔW are lightweight;

expensive for large # of workers
 Advantage: only need worker code (no central server code); if ΔW is
low rank, comms reduction possible

40

Halton Sequence Topology[Li et al., 2015]

 Used with decentralized storage paradigm
 Like P2P topology, but route messages through many workers
 e.g. to send message from 1 to 6, use 1->2->3->6

 Disadvantage: incur higher SSP staleness due to routing, e.g. 1->2->3->6 = staleness 3
 Advantage: support bigger messages; support more machines than
P2P topology

41

Random Partial Broadcastingand Diverse Mini-Batch Selection
 Random Partial Broadcasting Each machine randomly selects Q<<P machines to send messages(instead of full broadcast) Message cost reduced: from O(P2) to O(PQ), scales linearly withmachine count P!

 Diverse Mini-batch Selection
 Choose training data samples that maximize diversity score
 Pick few diverse samples instead of many random samples
 Using few but diverse samples further reduces comms costs without

hurting output quality!

42

Principles ofML system Design[Xing et al., to appear 2016]

4. What to Communicate:
Exploiting Structure in ML Updates

43

Matrix-Parameterized Models (MPMs)
Matrix parameter W

1 N
minf i (Wai ; bi ) + h(W )∑W
N i =1
Loss function

Regularizer

Distance Metric Learning, Sparse Coding, Distance Metric
Learning, Group Lasso, Neural Network, etc.

44

Billions of params = 10-100 GBs, costly
network synchronization

Big MPMs

What do we actually need to communicate?

Multiclass Logistic
Regression on Wikipedia
Feature dim. = 20K

6.5B

Feature dim. = 172K

#classes=325K

Sparse Coding on
ImageNet
Feature dim. = 172K

8.6B

Distance Metric Learningon ImageNet

8.6B

Latent dim. = 50K

Neural Network of
Google Brain#neurons in layer 0 = 40K

Dic. Size=50K

1.3B#neurons inlayer 1 = 33K
45

Full Updates
 Let matrix parameters be W. Need to send parallel worker

updates ΔW to other machines…
 Primal stochastic gradient descent (SGD)

1 N
min∑ fi (Wai ; bi ) + h(W )W
N i =1ΔW =∂f (Wai , bi )∂W Stochastic dual coordinate ascent (SDCA)

1 N *
* 1
min
f
(
−
z
)
+
h
( ZAT )∑iiZ
N i =1N

ΔW = (Δzi )ai46

Sufficient Factor (SF) Updates[Xie et al., 2015]

 Full parameter matrix update ΔW can be computed asouter product of two vectors uvT (called sufficient factors)
 Primal stochastic gradient descent (SGD)

1 N
min∑ fi (Wai ; bi ) + h(W )W
N i =1

ΔW = uv T u =∂f (Wai , bi )v = ai∂ (Wai ) Stochastic dual coordinate ascent (SDCA)

1 N *
* 1
min
f
(
−
z
)
+
h
( ZAT )∑iiZ
N i =1N

ΔW = uv T u = Δzi v = ai Send the lightweight SF updates (u,v), instead of the expensive

full-matrix ΔW updates!

47

P2P Topology + SF Updates= Sufficient Factor Broadcasting48

SFB Convergence Theorem[Xie et al., 2015]

Explanation: Parameter copies Wp on different workers p
converge to the same optima, i.e. all workers reach thesame (correct) answer.
üDoes not need central parameter server or key-value storeüWorks with SSP bridging model (staleness = s)49

Why is SFB faster?
 Faster than PS and Spark Near-linear scalability

50

Because SFB has faster iterations(less communication)… while keeping the same iteration quality as PS

SFB communication up to 100x smaller
than PS and SparkSummary
How to Distribute?

1.



Structure-Aware Parallelization
Work Prioritization

How to Bridge Computation and Communication?

2.



BSP Bridging Model
SSP Bridging Model for Data and Model Parallel

How to Communicate?

3.




Managed comms – interleave comms/compute, prioritized comms
Parameter Storage: Centralized vs Decentralized
Communication Topologies: Master-Slave, P2P, Halton Sequence

What to Communicate?

4.




Full Matrix updates
Sufficient Factor updates
Hybrid FM+SF updates (as in a DL model)
51