WIXLIB

Algorithm 1: MGD {T , η w0 , X}Algorithm 2: Distributed MGD {T , η w0 , X, m}Master:Issue LoadData() to all workers;Issue InitialModel(w0 ) to the central node;for Iteration t 0 to T doIssue WorkerTask(t) to all workers;Issue ServerTask(t) to the central node;for Iteration t 1 to T doSample a batch of data XB ;Compute gradient as gt xi XB l(xi , wt 1 );Update model as wt wt 1 η · gt η · Ω(wt 1 );Worker r 1, ., m:Function LoadData():Load a partition of data Xr ;Function WorkerTask(t):Get model wt 1 from the central node;if SendGradient thenSample a batch of data Xbr from Xr ; Compute gradient gtr xi Xbr l(wt 1 , xi );Send gradient gtr to the central node;else if SendModel thenCompute model wtr via MGD(T , η, wt 1 , Xr );// T is the number of iterations inside eachworker.Send local model wtr to the central node;The executions of GD and SGD (stochastic gradient descent [11], another popular variant of GD) are similar. Essentially, GD and SGD can be considered as special cases ofMGD. Specifically, when the batch size is the entire data (i.e.,XB X), it is GD; when the batch size is 1, it is SGD.Without loss of generality, we focus our discussion on MGD.B. Distributed MGDSequential execution of MGD is usually not feasible forlarge datasets and models. People have been proposing variousways of running MGD in a distributed manner. While thedetails of these proposals differ, we can use a generic architecture to capture the essence of these proposals, as presented inAlgorithm 2. In this generic architecture, there is a master topartition data and schedule tasks. There are multiple workers,each dealing with an individual partition. In addition, thereis a central node to aggregate the gradients/models receivedfrom the workers.As shown in Algorithm 2, the master first splits data intomultiple partitions. It then schedules each worker to loada partition and launch a training task. Each worker can beimplemented in two alternative ways: (SendGradient) Each worker pulls the latest modelfrom the central node. It then samples a batch from itslocal data, computes the gradient using the latest model,and sends the gradient to the central node. The centralnode aggregates the gradients received from the workersand updates the model. (SendModel) Each worker performs MGD over itslocal partition of the data and sends the updated (localview of the) model to the central node. The centralnode then updates the (global) model based on updatesreceived from the workers, using methods such as modelaveraging [9].The difference between the two paradigms lies in thenumber of updates to the (global) model within one singlecommunication step between the workers and the centralnode. If T 1, i.e., only one iteration is allowed inMGD, the number of updates made by SendGradient andSendModel will be exactly the same. However, if T 1,which is the typical case, SendModel will result in muchmore updates and thus much faster convergence.Central node:Function InitialModel(w0 ):Initialize model as w0 ;Function ServerTask(t):if SendGradient then mAggregate gradient as gt r 1 gtr ;Update the model aswt wt 1 η · (gt ) η · Ω(wt 1 );else if SendModel then mAggregate the models as wt f ( r 1 wtr );systems based on the parameter-server architecture (Petuumand Angel) from two aspects, i.e., (i) system architecture and(ii) algorithm implementation. We also briefly discuss otherrelevant systems, such as TensorFlow.A. Apache Spark and MLlibApache Spark [1] is a powerful framework for data analyticsthat has been adopted by enterprises across a wide range ofindustries, with attractive features such as fault tolerance andinteroperability with the Hadoop ecosystem. Spark can cachethe whole data in memory, so it fits well for iterative machinelearning workloads.MLlib [2] is one of the most popular machine learninglibraries built on top of Spark, which uses MGD to traingeneralized linear models. Conceptually, the execution ofMGD in MLlib can be outlined in Figure 2(a). There is adriver and multiple executors. The driver plays the role ofboth the master and the central node in Algorithm 2, which isresponsible for scheduling the executors and maintaining the(global) model. The executors serve as the workers in Algo-III. A NALYSIS OF E XISTING SYSTEMSIn this section, we provide detailed analysis of existingdistributed machine learning systems based on the generic architecture in Algorithm 2. We focus on anatomizing the implementation of MGD in Apache Spark (MLlib) and specialized1780

Broadcast modelExecutorExecutor " Local ModelLocal ModelLocal ModelExecutorExecutor % ! # & " ' # ! # " & % ! # ! ! ' Executor ' ' & & ( " " ! ! ( " % ! Driver * ! " ( ! Executor( ( & ( ( %& " ! " 0 ! # ! # ! ! " # " 0 !& ( # Executor ! # !! ( ExecutorExecutorLocal Model % ! % !! # # & & " " ! !! ' # # ! ! " & # ## # ! ' ' # ! % ! !Model Local " ! ! !!! ( " % & " " ( "" ! ( ! ! * %& " ! ( & # " 0 ! ( 0 !!& " & Local Model# ! ! # ! " "" ! ! # !# ! ! ( # ! # " (# (Local Model Local ModelExecutor ! ) ! (a) MLlib on Spark % ! # & " ! ' # ! " & % ! # ! ' & " ! ! ( " % ! " ( ! * ( %& " ( & # " 0 ! !! ! & # ! # ! " ! # ! ( # ( ! (b) AllReduce implementation using shuffle onSpark - . /(c) ML on Parameter ServerFig. 2. Communication patterns of machine learning on Spark and Parameter Server.rithm 2, which perform local computation over the partitioneddata. The execution of MGD leverages the SendGradientalternative: (1) The driver first broadcasts the current modelto the executors; (2) Each executor computes the gradientsusing its local data based on the received model, and sends thegradients to the driver; and (3) The driver aggregates gradientsharvested from the executors and updates the model.The driver can be overloaded when there are many executorsand the model is large. To alleviate the workload of the driver,MLlib implements the aggregation of gradients following ahierarchical, distributed style. As depicted in Figure 2(a), thedriver first employs some of the executors to perform localaggregation. It then pulls the aggregated (i.e., sum of) gradientsfrom these intermediate executors. This hierarchical dispatchof aggregation is called treeAggregate in MLlib.servers, and update the model using their local data; (2)Afterwards, workers send their local updates of the model toparameter servers; (3) Finally, the parameter servers sum upall the updates from the workers.The local computation that each worker of Petuum performsdepends on whether the regularization term in Equation 1 iszero or not — L2 regularization will result in dense updatesto the model that can be expensive [14]. If the regularizationterm is zero, Petuum workers instead conduct parallel SGDinside each batch; each communication step between workersand parameter servers therefore actually contains many localupdates to the model. If, on the other hand, there is a nonzeroregularization term, workers perform gradient descent over thebatch data in each iteration. Each communication step thuscontains only one update to the model. When updating themodel, the parameter servers sum up the model updates fromthe workers. We refer to this model aggregation technique asmodel summation in the rest of this paper.2) Angel: Angel [5] is another representative instance ofdistributed machine learning systems based on the parameterserver architecture, implemented in Java. Angel can read datadirectly from HDFS and run on Yarn clusters.Angel also adopts the SendModel paradigm to trainGLMs. It uses parameter servers to maintain the global modeland uses multiple workers to compute local updates of themodel. The difference between Angel and Petuum on trainingGLMs comes from two aspects. First, the frequency of communication is different. Workers in Angel communicate withthe parameter servers per epoch, whereas workers in Petuumcommunicate with the servers per batch. Second, the localcomputation performed on one batch of data is different. Angelalways performs gradient descent on each batch whereas theimplementation of Petuum depends on the regularization term,as we have described above.B. Parameter ServersFigure 2(c) presents a typical parameter-server architecture [12] based machine learning system. Unlike MLlib wherea single driver is responsible for maintaining the (global)model, in parameter-server architecture the model is storedacross multiple machines called parameter servers. In otherwords, there are multiple nodes that serve the role of thecentral node in Algorithm 2. Moreover, unlike the BulkSynchronous Parallel (BSP) execution model that backs upthe implementation of Spark, workers can communicate withparameter servers asynchronously. Parameter servers can leverage different consistency controllers to implement differentcommunication schemes such as BSP, SSP (acronym for “StaleSynchronous Parallel”), and ASP (acronym for “AsynchronousParallel”), by enabling or disabling requests from workers.It has been shown that asynchronous communication can bebeneficial for distributed machine learning [13].In the following, we briefly review two instances of theparameter-server architecture: Petuum [7] and Angel [5].1) Petuum: Petuum is one of the state-of-the-art distributedmachine learning systems with the parameter-server architecture, implemented in C . Unlike MLlib that followsthe SendGradient paradigm, Petuum instead adopts theSendModel alternative in Algorithm 2 to train GLMs. Thatis, the workers in Petuum will update local model immediatelyafter computing the gradients and send the updates to theparameter servers. In more detail, the workflow for eachiteration is: (1) Workers pull the latest model from parameterC. Other SystemsClearly, MGD has been implemented in many other distributed machine learning systems, in particular systems developed for deep learning, such as TensorFlow, as GD isessential for the backpropagation algorithm that trains deepneural networks. However, we decide to not include thesesystems in this study because it has been pointed out inthe literature that TensorFlow is slower than Petuum andAngel when training GLMs because of too many abstractions,1781

which will significantly raise system complexity as well asruntime overhead [5], [6]. Furthermore, TensorFlow does notprovide mechanism for partitioning models. This also leads toinefficiency when handling large models in GLMs.the intermediate aggregators is also understandable dueto the hierarchical aggregation mechanism employed byMLlib, although it shifts some workload from the driver— the latency at the driver can be even worse without thishierarchical scheme.IV. ML LIB *: U NDERSTANDING AND I MPROVINGP ERFORMANCE OF ML LIBB. Implementation of MLlib*Based on our analysis in the previous sections, the implementation of MGD in MLlib is clearly not optimal. By usingthe SendGradient paradigm instead of the SendModelparadigm that has been widely implemented in, e.g., parameter servers, MLlib is likely to suffer from a much slowerconvergence within the same number of communication steps.This inferior performance has been noticeably documented inthe literature [6]. A natural follow-up question is whether thisapparently flawed implementation can be improved within theBSP framework that Spark is based on.In this section, we identify performance bottlenecks of MLlib and study state-of-the-art, well-known techniques that cansignificantly improve the performance of MLlib when runningMGD. Our implementations of these techniques piggyback onthe existing Spark primitives and only require minor changesto the current MLlib implementation.In MLlib*, we use two well known techniques to deal withthe two bottlenecks in MLlib’s implementation: (1) modelaveraging and (2) distributed aggregation. We now discussthem in detail.1) Model Averaging: Model averaging is the essentialreason for the efficiency of the SendModel paradigm inAlgorithm 2. If we can implement model averaging in MLlib,the driver will remain as a bottleneck but much less frequently,due to the reduced number of communication steps betweenthe driver and the executors.Most part of our implementation is quite straightforward.We basically replace the computation of gradients in eachexecutor by model updates, and change the data being sentfrom gradients to model updates, too. However, SendModelcan be inefficient when the regularization term (typically L2norm) is not zero. In this case, frequent updates to the localview of the model can be quite expensive when the modelsize is large. To address this, we use a threshold-based,lazy method to update the models following Bottou [14].Our implementation does not require any change to the coreof Spark. Instead, we implement our techniques leveragingprimitives provided by Spark.Figure 3(b) presents the gantt chart of MLlib after incorporating our implementation of SendModel, by rerunningthe experiment described in Section IV-A. One can observethat it is very similar to Figure 3(a). It implies that ourimplementation does not impact per-node computation timemuch. This is not surprising, as the computational tasks ofSendModel are similar to those of SendGradient — itis just computing weights of the model versus computinggradients! Nonetheless, the number of stages in Figure 3(b)should be much smaller comparing with Figure 3(a) if weextend the x-axes of both charts to the time of convergence,which suggests a much faster convergence of SendModel.Remark: Note that model averaging is just one ofthe model aggregation schemes that can be adopted in theSendModel paradigm. For example, Petuum instead leverages the model summation scheme, which simply sums upbut does not take the average of the received model updates.As was pointed out by Zhang and Jordan [15], there are prosand cons between model averaging and model summation.Roughly speaking, model summation can lead to potentialdivergence; however, when it converges, it can converge fasterthan model averaging. They actually proposed a technique thatfurther improves the convergence rate of model averaging by“reweighting” the samples taken when combining the modelupdates. Implementing their techniques may therefore furtherimprove the performance of MLlib*. However, as we willsee, even with the current implementation, the performance ofA. Bottlenecks in MLlibWe start by giving a more detailed analysis to understandbottlenecks in MLlib. We ran MGD to train a linear supportvector machine (SVM) using the kdd12 dataset describedin Table I. The experiment was conducted on a cluster ofnine nodes with one node serving as the driver and theothers serving as the executors in Spark (recall Figure 2(a)). 1Figure 3(a) presents the gantt chart2 that tracks the executionof the nodes. The x-axis represents the elapsed time (inseconds) since the start of the execution. The y-axis representsthe activities of the driver and the eight executors as timegoes by. Each colored bar in the gantt chart represents atype of activity during that time span that is executed in thecorresponding cluster node (i.e., driver or executor), whereasdifferent colors represent different types of activities.We can identify two obvious performance issues by examining the gantt chart in Figure 3(a): (B1) Bottleneck at the driver — at every stage when thedriver is executing, the executors have to wait. (B2) Bottleneck at the intermediate aggregators, i.e.,the executors performing intermediate aggregations ofgradients — at every stage when these executors arerunning, the other nodes have to wait.As discussed in Section III-A, MLlib uses theSendGradient paradigm. The bottleneck at the driveris therefore easy to understand: the executors simplycannot proceed because they have to wait for the driverto finish updating the model. Moreover, the bottleneck at1 We assign one task to each executor because when we increase the numberof tasks per executor, the time per iteration increases due to the heavycommunication overhead.2 https://en.wikipedia.org/wiki/Gantt chart1782

Executor 1Executor 2Executor 3Executor 4Executor 5Executor 6Executor 7Executor 8Driver0Executor 1Executor 2Executor 3Executor 4Executor 5Executor 6Executor 7Executor 8DriverExecutor 1Executor 2Executor 3Executor 4Executor 5Executor 6Executor 7Executor 8Driver50100150200250300050Time (s)(a) MLlib150100200250300050100150200Time (s)Time (s)(b) MLlib model averaging(c) ML

extracted and transformed using Spark. However, the commonly used machine learning systems are TensorFlow, XGBoost, and Angel, whereas Spark MLlib, an official Spark package for machine learning, is seldom used. One reason for this ignorance is that it is generally believed that Spark is slow when it comes to distributed machine learning.