We present the rst protocol for distributed online prediction that aims to minimize online prediction loss and network communication at the same time. Applications include social content recommendation, algorithmic trading, and other scenarios where a con guration of local prediction models of high-frequency streams is used to provide a realtime service. For stationary data, the proposed protocol retains the asymptotic optimal regret of previous algorithms. At the same time, it allows to substantially reduce network communication, and, in contrast to previous approaches, it remains applicable when the data is non-stationary and shows rapid concept drift. The protocol is based on controlling the divergence of the local models in a decentralized way. Its bene cial properties are also con rmed empirically.
We consider online prediction problems where data points are observed at local nodes in a distributed environment and there is a trade-o between maximizing prediction accuracy and minimizing network communication. This situation abounds in a wide range of machine learning applications, in which communication induces a severe cost. Examples are parallel data mining [Zinkevich et al., 2009, Hsu et al.] where communication constitutes a performance bottleneck, learning with mobile sensors [Nguyen et al., 2004, Predd et al., 2006] where communication drains battery power, and, most centrally, prediction-based real-time services [Dekel et al., 2012] carried out by several servers, e.g., for social content promotion, ad placement, or algorithmic trading. In addition to the above, here the cost of communication can also be a loss of prediction quality itself when training examples have to be discarded due to network latency. 1.1
Despite the communication costs it induces,
decentralization is inevitable in many modern scale applications. Hence,
recent articles [Balcan et al., 2012, Daume III et al.]
explicitly investigate the communication complexity of learning
with decentralized data. They consider, however, the
ofine task of nding a good global model over the union of
all data as a nal computation result. The same applies
to some work on parallel machine learning
To this end, research focused so far on speci c environments with xed communication constraints. Correspondingly, the learning strategies that are proposed and analyzed for these settings, do not aim to minimize communication beyond the level that is enforced by these constraints. Zinkevich et al. [2009] considers a shared-memory model, in which all local nodes can update a global model in a round-robin fashion as they process their training examples. Since this approach is problematic if there is a notable communication latency, strategies have been investigated [Mann et al., 2009, Dekel et al., 2012] that communicate only periodically after a statically xed number of data points have been processed. Dekel et al. [2012] shows that for smooth loss functions and stationary environments optimal asymptotic regret bounds can be retained by updating a global model only after minibatches of O( p3m) data points. Here, m denotes the total number of data points observed throughout the lifetime of the system. For large values of m, the e ect of bounded latency values is asymptotically outgrown by the increasing mini-batch size.
While a xed periodic communication schedule reduces the communication by some xed amount, further reduction is desirable: The above mentioned costs of communication can have a severe impact on the practical performance|even if they are not re ected in asymptotic performance bounds. This is further ampli ed because a large number of modeling tasks are performed simultaneously sharing the same limited bandwidth. Moreover, distributed learning systems that are deployed for a long lifetime relative to their data throughput can experience periodical or singular target drifts (e.g., corresponding to micro-trends in social networks). In these settings, a static schedule is bound to either provide only little to no communication reduction or to insu ciently react to changing data distributions. 1.2
In this work, we give the rst distributed prediction protocol for linear models that, at the same time, aims to provide a high online in-place prediction performance and explicitly tries to minimize communication. In terms of predictive power, as shown Sec. 3.1, the protocol retains the asymptotic optimal regret of the distributed mini-batch algorithm of Dekel et al. [2012] for stationary data. In addition, it allows to reduce the communication among the local nodes substantially. This is achieved by a dynamic data dependent communication schedule, which, in contrast to previous algorithms, remains applicable when the data is nonstationary and shows rapid concept drifts. The main idea is to synchronize the local models to their mean model in order to reduce their variance, but to do so only in system states that show a high divergence among the models. This divergence, measured by the average model distance to the mean model, indicates the synchronizations that are most important in terms of their correcting e ects on the predictions. In stable phases this allows communicative quiescence, while, in hard phases where variance reduction is crucial, the protocol will trigger a lot of model synchronizations. In order to e ciently implement this strategy one has to monitor the non-linear divergence function without communication overhead. We propose a solution to this problem that adapts recent ideas from distributed systems research based on local safe-zones in the function domain (Sec. 3.2). Experiments con rm the bene cial properties of the protocol (Sec. 4).
In this section we formally introduce the distributed online prediction task. As simple local learning tool we recall stochastic gradient descent for linear models. Finally, we review the state-of-the-art communication protocol as a departure point for developing a more communication-e cient solution in subsequent sections. 2.1
Throughout this paper we consider a distributed online prediction system of k local learners that maintain individual linear models wt;1; : : : ; wt;k 2 Rn of some global environment through discrete time t 2 [T ] where T 2 N denotes the total time horizon with respect to which we analyze the system's performance. This environment is represented by a target distribution Dt : X Y ! [0; 1] that describes the relation between an input space X Rn and an output space Y R. The nature of Y varies with the learning task at hand; Y = f 1; 1g is used for binary classi cation, Y = R for regression. While we allow Dt to vary with time, we assume that it remains constant most of the time and only experiences a small number of rapid drifts. That is, there are drift points 0 = d0 < d1 < < dp = T such that for all i 2 [p] and t; t0 2 [T ] with di 1 t t0 < di it holds that Dt = Dt0 . Hence, there are identically distributed episodes
Ei = fdi; : : : ; di+1 1g between any two drift points. We assume that all learners sample from D independently in parallel using a constant and uniform sampling frequency, and we denote by (xt;l; yt;l) Dt the training example received at node l at time t. Generally, we assume that all training examples are bounded by a ball with radius R.
Conceptually, every learner rst observes the input part xt;l and performs a real time service based on the linear prediction score pt;l = hwt;l; xt;li, i.e., the inner product of xt;l and the learner's current model vector. Only then it receives as feedback the true label yt;l, which it can use to locally update its model to wt+1;l = '(wt;l; xt;l; yt;l) by some update rule ' : Rn X Y ! Rn. Finally, the learners are connected by a communication infrastructure that allows them to jointly perform a synchronization operation : Rk n ! Rk n that resets the whole model con guration to a new state and that may take into account the information of all local learners simultaneously. The performance of such a distributed online prediction system is measured by two quantities: 1) the predictive performance PtT=1 Plk=1 f (pt;l; yt;l) measured by a loss function f : R Y ! R+ that assigns positive penalties to prediction scores; and 2) the amount of communication within the system that is measured by the number of bits sent inbetween learners to compute the sync operation. Next, specify possible choices for the update rule, the loss function, and the synchronization operator. 2.2
Generally, the communication protocol developed in this paper is applicable to a wide range of online update rules for linear models from, e.g., the passive aggressive rule [Crammer and Singer, 2001] to regularized dual averaging [Xiao, 2010]. However, the regret bound given in Theorem 2 assumes that the updates are contractions. That is, there is some constant c < 1 such that for all w; w0 2 Rn, and x; y 2 X Y it holds that k'(w; x; y) '(w0; x; y)k ckw w0k. For the sake of simplicity, in this paper, we focus on rules based on l2-regularized stochastic gradient descent, for which this contraction property is readily available. We note that by considering expected contractions the result can be extended to rules that reduce on average the distance to a (regularized) loss minimizer.
Before we can de ne gradient descent updates, we have to introduce the underlying loss functions measuring predictive performance. Again for convenience, we restrict ourselves to functions that are di erentiable, convex, and globally Lipschitz continuous in the prediction score, i.e., there is some constant L such that for all p; p0; y 2 R2n Y it holds that jf (p; y) f (p0; y)j Ljp p0j. While these assumptions can be relaxed by spending some technical e ort, they already include loss functions for all standard predictions tasks such as the logistic loss flg(p; y) = ln(1 + exp( yp)) for binary classi cation (case Y = f 1; 1g) or the Huber loss for regression (in the case Y = R) fhu(p; y) = ( 21 (p jp
y)2 yj
With this we can de ne stochastic gradient descent (SGD) rules with l2-regularization, i.e., rules of the form '(w; x; y) = w trw 2 kwk2 + f (hw; xi; y) where 2 R+ is a strictly positive regularization parameter and t 2 R+ are strictly positive learning rates for t 2 N. For stationary target distributions, one often chooses a decreasing learning rate such as t = 1=pt in order to guarantee convergence of the learning process. For non-stationary targets this is infeasible, because for large t it would prevent su cient model adaption to target changes. However, one can show [Zinkevich et al., 2010] that stochastic gradient descent is a contraction for su ciently small constant learning rates. Namely, for (RL + ) 1 the updates do contract with constant c = 1 . This can be used to show that the stochastic learning process converges to a distribution centered close to a regularized loss minimizer even when the process is distributed among k nodes (see the analysis of Zinkevich et al. [2010]). This refers to the stochastic learning process de ned by the mean of independent local models that result from SGD with iid samples from (episodes of) the target distribution. In this paper, the contraction property is used for the regret bound of Thm. 2. 2.3
For every episode Ei, the predictive performance of a
distributed prediction system lies between two baselines that
correspond to the two extremes in terms of communication
behavior|complete centralization and no communication.
Let Ti = jEij denote the length of episode Ei and by R =
Pt2Ei; l2[k] f (pt;l; yt;l) f the regret with respect to the
optimal expected loss f = argminw2Rn E(x;y) Di [f (hw; xi; y)].
When all data points are centrally processed by one online
learner, for long enough episodes one can achieve an
expected regret of O(pkTi) which is optimal
Mann et al. [2009] and Dekel et al. [2012] give algorithms
where information between nodes is only exchanged every
b rounds where b 2 N is referred to as batch size. These
algorithms can be written as static model synchronization
protocol similar to Alg. 1. Here, after a batch of kb examples
has been processed globally in the system, all local models
are re-set to the mean model of the con guration w
dened as w = 1=k Plk=1 wl. Formally, the synchronization
operator that is implicitly employed in these algorithms is
given by (wt) = (wt; : : : ; wt). We refer to this operation
as full mean synchronization. The choice of a (uniform)
model mixture is often used for combining linear models that
have been learned in parallel on independent training data
Closing this section, let us analyze the communication cost of this protocol.Using a designated coordinator note as in Alg. 1, can computed simply by all nodes sending their current model to the coordinator, who in turn computes the mean model and sends it back to all the nodes. For assessing the communication cost of this operation, we only count the number of model vectors sent between the learners. This is feasible because, independently of the exact communication infrastructure, the number of model messages asymptotically determines the true bit-based cost. Hence, asymptotically the communication cost of static model synchronization over k nodes with batch size b is O(kT =b). Dekel et al. [2012] assumes that the data distribution is stationp ary over all rounds and b can therefore be set to O( 3 T ). This results in an automatic communication reduction that increases with a longer system lifetime. However, this strategy is not applicable when we want to stay adaptive towards changing data distributions. In this case, we have to set the batch size with respect to the expected episode length and not with respect to the overall system lifetime. This number can be much smaller than T resulting in batch sizes that are too small to meet our communication reduction goal. In the following section, we therefore design a synchronization protocol that can substantially reduce this cost based on a data-dependent dynamic schedule.
The synchronization protocol of Alg. 1 is static because it synchronizes after a xed number of rounds independently of the sampled data and its e ect on the local models. Consequently, it incurs the communication cost of a full synchronization round even if the models are (almost) identical and thus only receive little to none correction. In this section, we develop a dynamic protocol for synchronizations based on quantifying their e ect. After showing that this approach is sound from a learning perspective, we discuss how it can be implemented in a communication-e cient way. 3.1
A simple measure to quantify the correcting e ect of synchronizations is given by the average Euclidean distance between the current local models and the result model. We refer to this quantity as the divergence of a model conguration, denoted by ( ), i.e., (w) = k1 Plk=1 kw wlk2. In the following de nition we provide a relaxation of the full mean synchronization operation that introduces some leeway in terms of this divergence.
Definition 1. A partial synchronization operator with a positive divergence threshold 2 R is an operator : Rk n ! Rk n that 1) leaves the mean model invariant and 2) after its application the model divergence is bounded by . That is, for all model con gurations w 2 Rk n it holds that w = w and ( w) .
An operator adhering to this de nition does not generally put all nodes into sync (albeit the fact that we still refer to it as synchronization operator). In particular it allows to leave all models untouched as long as the divergence remains below the threshold . The following theorem notes that partial synchronization has a controlled regret over full synchronization if the batch size is su ciently large and the divergence threshold is set proportional to the Lipschitz constant L of the losses and the data radius R.
Theorem 2. Suppose the update rule ' is a contraction with constant c. Then, for batch sizes b log2 1 c 1 and divergence thresholds =(2RL), the average regret of using a partial synchronization operator instead of is bounded by , i.e., for all rounds t 2 N it holds that the average regret 1=k Plk=1 jf (pt;l; yt;l) f (pt;l; yt;l)j is bounded by where pt;l and pt;l denote the prediction scores at learner l and time t resulting from and , respectively. We omit the proof here referring to the full version of this paper. While the contraction assumption is readily available for regularized SGD, as mentioned in Sec. 2, it can be relaxed: by requiring the updates to only contract on expectation it is possible to extend the theorem to unregularized SGD updates as well as to other rules. Moreover, we remark that Thm. 2 implies that partial synchronizations retain the optimality of the static mini-batch algorithm of Dekel et al. [2012] for the case of stationary targets: By using a timedependent divergence threshold based on t 2 O(1=pt) the bound of O(pT ) follows. 3.2
After seeing that partial synchronization operators are
sound from the learning perspective, we now turn to how
they can be implemented in a communication-e cient way.
Every distributed learning protocol that implements a
partial synchronization operator has to implicitly control the
divergence of the model con guration. However, we cannot
simply compute the divergence by centralizing all local
models, because this would incur just as much communication
as static full synchronization. Our strategy to overcome this
problem is to rst decompose the global condition (w)
into a set of local conditions that can be monitored at their
respective nodes without communication
local models w1;1; : : : ; w1;k reference point r (0; : : : ; 0) violation counter v 0 At coordinator on violation: let B be set of nodes with violation v v + jBj if v = k then B [k], v 0 while B 6= [k] and kr B1 Pl2B wlk > do augment B by augmentation strategy receive models from nodes added to B send to nodes in B model w = B1 P if B = [k] also set new reference modl2elBrwl w that transfers the system back into a valid state whenever one or more of these local conditions are violated. This includes carrying out a su cient amount of synchronization to reduce the divergence to be less or equal than .
For deriving local conditions we consider the domain of
the divergence function restricted to an individual model
vector. Here, we identify a safe-zone S
Theorem 3. Let r 2 Rd be some reference point. If for all nodes l 2 f1; : : : ; kg it holds that kr wlk =2 then we have for the model divergence that (w) .
We now incorporate these local conditions into a distributed prediction protocol. As a rst step, we have to guarantee that at all times all nodes use the same reference point. For a prediction t, let us denote by t0 the last time prior to t when a full model synchronization was performed (resp. t0 = 0 in case no full synchronization has happened until round t). The mean model wt0 is known to all local learners. We use this model as the reference model and set r = wt0 . A local learners l can then monitor their local condition kr wlk =2 in a decentralized manner.
It remains to design a resolution protocol that speci es
how to react when one or several of the local conditions are
violated. A direct solution is to trigger a full
synchronization in that case. This approach, however, does not scale
well with a high number of nodes in cases where model
updates have a non-zero probability even in the asymptotic
regime of the learning process. When, e.g., PAC models for
the current target distribution are present at all local nodes,
the probability of one local violation, albeit very low for an
individual node, increases exponentially with the number of
nodes. An alternative approach that can keep the amount
1Note that a direct distribution of the threshold across the
local nodes
No Synchronization static (batch sizes) dynamic (div. thres.) 0.05 150000 200000 Number of messages of communication low relative to the number of nodes is to perform a local balancing procedure: on a violation, the respective node sends his model to a designated note we refer to as coordinator. The coordinator then tries to balance this violation by incrementally querying other nodes for their models. If the mean of all received models lies within the safe zone, it is transferred back as new model to all participating nodes, and the resolution is nished. If all nodes have been queried, the result is equal to a full synchronization and the reference point can be updated. In both cases, the divergence of the model con guration is bounded by at the end of the balancing process, because all local conditions hold. Also this protocol leaves the global mean model unchanged. Hence, it is complying to Def. 1.
While balancing can achieve a high communication reduction over direct resolution particularly for a large number of nodes, it potentially degenerates in certain special situations: We can end up in a stable regime in which local violations are likely to be balanced by a subset of the nodes; however a full synchronization would strongly reduce the expected number of violations in future rounds. In other words: balancing can delay crucial reference point updates inde nitely. A simple hedging mechanism for online optimization can be employed to avoid this situation: we count the number of local violations using the current reference point and trigger a full synchronization whenever this number exceeds the number of nodes. This concludes our dynamic protocol for distributed prediction. All components are summarized in Alg. 2
In this section we investigate the practical performance of the dynamic learning protocol for two controlled settings: one with linearly separable data and one with unseparable data. Our main goal is to empirically con rm that the predictive gain of static full synchronizations (using a batch size of 8) over no synchronization can be approximately preserved for small enough thresholds, and to assess the amount of communication reduction achieved by these thresholds.
We start with the problem of tracking a rapidly drifting random disjunction. In this case the target distribution produces data that is episode-wise linearly separable. Hence, we can set up the individual learning processes so that they converge to a linear model with zero classi cation error within each episode. Formally, we identify a target disjunction with a binary vector z 2 f0; 1gn. A data point x 2 X = f0; 1gn is labeled positively y = 1 if hx; zi 1 and otherwise receives a negative label y = 1. The target disjunction is drawn randomly at the beginning of the learning process and is randomly re-set after each round with a xed drift probability of 0:0002. In order to have balanced classes, the disjunctions as well as the data points are generated such that each coordinate is set independently to 1 with probability p1 2 1=n. As loss function for the stochastic gradient descent we use the logistic loss. Corresponding to our setting of noise-free linearly separable data, we choose the regularization parameter = 0 and the learning rate = 1.
In Fig. 1 (left) we present the result for dimensionality n = 100, with k = 512 nodes, processing m = 12:8M data points through T = 25000 rounds. For divergence thresholds up to 3:0, dynamic synchronization can retain the error number of statically synchronizing every 8 rounds. At the same time the communication is reduced to 3:9% of the original number of messages. An approximately similar amount of communication reduction can also be achieved using static synchronization by increasing the batch size to 128. This approach, however, only retains 51:5% of the error reduction over no communication. Analyzing the development of the evaluation metrics over time reveals: At the beginning of each episode there is a relatively short phase in which additional errors are accumulated and the communicative protocols acquire an advantage over the baseline of never synchronizing. This is followed by a phase during which no additional error is made. Here, the communication curve of the dynamic protocols remain constant acquiring a gain over the static protocols in terms of communication.
We now turn to a harder experimental setting, in which
the target distribution is given by a rapidly drifting
twolayer neural network. For this target even the Bayes
optimal classi er per episode has a non-zero error, and, in
particular, the generated data is not linearly separable.
Intuitively, it is harder in this setting to save communication,
because a non-zero residual error can cause the linear
models to periodically uctuate around a local loss minimizer|
resulting in crossings of the divergence threshold even when
the learning processes have reached their asymptotic regime.
We choose the network structure and parameter ranges in
a way that allow for a relatively good approximation by
linear models
Figure 1 (right) contains the results for dimensionality 150, with k = 1024 nodes, processing m = 2:56M data points through T = 2500 rounds. For divergence thresholds up to 0:05, dynamic synchronization can retain the error of the baseline. At the same time the communication is reduced to 46% of the original number of messages.
We presented a protocol for distributed online prediction that aims to dynamically save on network communications in su ciently easy phases of the modeling task. The protocol has a controlled predictive regret over its static counterpart and experiments show that it can indeed reduce the communication substantially|up to 95% in settings where the linear learning processes are suitable to model the data well and converge reasonably fast. Generally, the e ectivity of the approach appears to correspond to the e ectivity of linear modeling by SGD in the given setting.
For future research a theoretical characterization of this behavior is desirable. A practically even more important direction is to extend the approach to other model classes that can tackle a wider range of learning problems. In principle, the approach of controlling model divergence remains applicable, as long as the divergence is measured with respect to a distance function that induces a useful loss bound between two models. For probabilistic models this can for instance be the KL-divergence. However, more complex distance functions constitute more challenging distributed monitoring tasks, which currently are open problems.
Martin Zinkevich, Alex J. Smola, and John Langford. Slow learners are fast. In Proc. of 23rd Annual Conference on Neural Information Processing Systems (NIPS 2009), pages 2331{2339, 2009.