Research on distributed machine learning algorithms has focused primarily on one of two extremes-algorithms that obey strict concurrency constraints or algorithms that obey few or no such constraints. We consider an intermediate alternative in which algorithms optimistically assume that conflicts are unlikely and if conflicts do arise a conflict-resolution protocol is invoked. We view this "optimistic concurrency control" paradigm as particularly appropriate for large-scale machine learning algorithms, particularly in the unsupervised setting. We demonstrate our approach in three problem areas: clustering, feature learning and online facility location. We evaluate our methods via large-scale experiments in a cluster computing environment. The desire to apply machine learning to increasingly larger datasets has pushed the machine learning community to address the challenges of distributed algorithm design: partitioning and coordinating computation across the processing resources. In many cases, when computing statistics of iid data or transforming features, the computation factors according to the data and coordination is only required during aggregation. For these embarrassingly parallel tasks, the machine learning community has embraced the map-reduce paradigm, which provides a template for constructing distributed algorithms that are fault tolerant, scalable, and easy to study. However, in pursuit of richer models, we often introduce statistical dependencies that require more sophisticated algorithms (e.g., collapsed Gibbs sampling or coordinate ascent) which were developed and studied in the serial setting. Because these algorithms iteratively transform a global state, parallelization can be challenging and often requires frequent and complex coordination. Recent efforts to distribute these algorithms can be divided into two primary approaches. The mutual exclusion approach, adopted by [12] and [16] , 1 arXiv:1307.8049v1 [cs.LG] 30 Jul 2013 guarantees a serializable execution preserving the theoretical properties of the serial algorithm but at the expense of parallelism and costly locking overhead. Alternatively, in the coordination-free approach, proposed by [21] and [1] , processors communicate frequently without coordination minimizing the cost of contention but leading to stochasticity, data-corruption, and requiring potentially complex analysis to prove algorithm correctness. In this paper we explore a third approach, optimistic concurrency control (OCC) [14] which offers the performance gains of the coordination-free approach while at the same time ensuring a serializable execution and preserving the theoretical properties of the serial algorithm. Like the coordination-free approach, OCC exploits the infrequency of data-corrupting operations. However, instead of allowing occasional data-corruption, OCC detects data-corrupting operations and applies correcting computation. As a consequence, OCC automatically ensures correctness, and the analysis is only necessary to guarantee optimal scaling performance. We apply OCC to distributed nonparametric unsupervised learningincluding but not limited to clustering-and implement distributed versions of the DP-Means We demonstrate how to analyze OCC in the context of the DP-Means algorithm and evaluate the empirical scalability of the OCC approach on all three of the proposed algorithms. The primary contributions of this paper are: 1. Concurrency control approach to distributing unsupervised learning algorithms. 2. Reinterpretation of online nonparametric clustering in the form of facility location with approximation guarantees. 3. Analysis of optimistic concurrency control for unsupervised learning. 4. Application to feature modeling and clustering. Many machine learning algorithms iteratively transform some global state (e.g., model parameters or variable assignment) giving the illusion of serial dependencies between each operation. However, due to sparsity, exchangeability, and other symmetries, it is often the case that many, but not all, of the state-transforming operations can be computed concurrently while still preserving serializability: the equivalence to some serial execution where individual operations have been reordered. This opportunity for serializable concurrency forms the foundation of distributed database systems. For example, two customers may concurrently make purchases exhausting the inventory of unrelated products, but if they try to purchase the same product then we may need to serialize their purchases to ensure sufficient inventory. One solution (mutual exclusion) associates locks with each product type and forces each purchase of the same product to be processed serially. This might work for an unpopular, rare product but if we are 2 interested in selling a popular product for which we have a large inventory the serialization overhead could lead to unnecessarily slow response times. To address this problem, the database community has adopted optimistic concurrency control (OCC) [14] in which the system tries to satisfy the customers requests without locking and corrects transactions that could lead to negative inventory (e.g., by forcing the customer to checkout again). Optimistic concurrency control exploits situations where most operations can execute concurrently without conflicting or violating serial invariants in our program. For example, given sufficient inventory the order in which customers are satisfied is immaterial and concurrent operations can be executed serially to yield the same final result. However, in the rare event that inventory is nearly depleted two concurrent purchases may not be serializable since the inventory can never be negative. By shifting the cost of concurrency control to rare events we can admit more costly concurrency control mechanisms (e.g., re-computation) in exchange for an efficient, simple, coordination-free execution for the majority of the events. Formally, to apply OCC we must define a set of transactions (i.e., operations or collections of operations), a mechanism to detect when a transaction violates serialization invariants (i.e., cannot be executed concurrently), and a method to correct (e.g., rollback) transactions that violate the serialization invariants. Optimistic concurrency control is most effective when the cost of validating concurrent transactions is small and conflicts occur infrequently. Machine learning algorithms are ideal for optimistic concurrency control. The conditional independence structure and sparsity in our models and data often leads to sparse parameter updates substantially reducing the chance of conflicts. Similarly, symmetry in our models often provides the flexibility to reorder serial operations while preserving algorithm invariants. Because the dependency structure is encoded in the model we can easily detect when an operation violates serial invariants and correct by rejecting the change and rerunning the computation. Alternatively, we can exploit the semantics of the operations to resolve the conflict by accepting a modified update. As a consequence OCC allows us to easily construct provably correct and efficient distributed algorithms without the need to develop new theoretical tools to analyze chaotic convergence or non-deterministic distributed behavior. Optimistic concurrency control can be distilled to a simple pattern (metaalgorithm) for the design and implementation of distributed machine learning systems. We begin by evenly partitioning N data points (and the corresponding computation) across the P available processors. Each processor maintains a replicated view of the global state and serially applies the learning algorithm as a sequence of operations on its assigned data and the global state. If an operation mutates the global state in a way that preserves the serialization invariants then the operation is accepted locally and its effect on the global state, if any, is eventually replicated to other processors. However, if an operation could potentially conflict with operations on other processors then it is sent to a unique serializing processor where it is rejected or corrected and the resulting global state change is eventually replicated to the rest of the processors. Meanwhile the originating processor either tentatively accepts the state change (if a rollback operator is defined) or proceeds as though the operation has been deferred to some point in the future. While it is possible to execute this pattern asynchronously with minimal coordination, for simplicity we adopt the bulk-synchronous model of [23] and divide the computation into epochs. Within an epoch t, b data points B(p, t) are evenly assigned to each of the P processors. Any state changes or serialization operations are transmitted at the end of the epoch and processed before the next epoch. While potentially slower than an asynchronous execution, the bulksynchronous execution is deterministic and can be easily expressed using existing systems like Hadoop or Spark [25] . Much of the existing literature on distributed machine learning algorithms has focused on classification and regression problems, where the underlying model is continuous. In this paper we apply the OCC pattern to machine learning problems that have a more discrete, combinatorial flavor-in particular unsupervised clustering and latent feature learning problems. These problems exhibit symmetry via their invariance to both data permutation and cluster or feature permutation. Together with the sparsity of interacting operations in their existing serial algorithms, these problems offer a unique opportunity to develop OCC algorithms. The K-means algorithm provides a paradigm example; here the inferential goal is to partition the data. Rather than focusing solely on K-means, however, we have been inspired by recent work in which a general family of K-means-like algorithms have been obtained by taking Bayesian nonparametric (BNP) models based on combinatorial stochastic processes such as the Dirichlet process, the beta process, and hierarchical versions of these processes, and subjecting them to small-variance asymptotics where the posterior probability under the BNP model is transformed into a cost function that can be optimized [4] . The algorithms considered to date in this literature have been developed and analyzed in the serial setting; our goal is to explore distributed algorithms for optimizing these cost functions that preserve the structure and analysis of their serial counterparts. We first consider the DP-means algorithm (Alg. 1) introduced by [13] . Like the K-means algorithm, DP-Means alternates between updating the cluster assignment z i for each point x i and recomputing the centroids C = {µ k } K k=1 associated with each clusters. However, DP-Means differs in that the number of clusters is not fixed a priori. Instead, if the distance from a given data point to 4 Algorithm 1: Serial DP-means Output: Accepted cluster centers C Algorithm 2: DPValidate Input: Set of proposed cluster centerŝ Output: Accepted cluster centers C Figure 1 : The Serial DP-Means algorithm and distributed implementation using the OCC pattern. 5 all existing cluster centroids is greater than a parameter λ, then a new cluster is created. While the second phase is trivially parallel, the process of introducing clusters in the first phase is inherently serial. However, clusters tend to be introduced infrequently, and thus DP-Means provides an opportunity for OCC. In Alg. 3 we present an OCC parallelization of the DP-Means algorithm in which each iteration of the serial DP-Means algorithm is divided into N/(P b) bulk-synchronous epochs. The data is evenly partitioned {x i } i∈B(p,t) across processor-epochs into blocks of size b = |B(p, t)|. During each epoch t, each processor p evaluates the cluster membership of its assigned data {x i } i∈B(p,t) using the cluster centers C from the previous epoch and optimistically proposes a new set of cluster centersĈ. At the end of each epoch the proposed cluster centers,Ĉ, are serially validated using Alg. 2. The validation process accepts cluster centers that are not covered by (i.e., not within λ of) already accepted cluster centers. When a cluster center is rejected we update its reference to point to the already accepted center, thereby correcting the original point assignment. The DP-Means objective turns out to be equivalent to the classic Facility Location (FL) objective: which selects the set of cluster centers (facilities) µ ∈ C that minimizes the shortest distance x − µ to each point (customer) x as well as the penalized cost of the clusters λ 2 |C|. However, while DP-Means allows the clusters to be arbitrary points (e.g., C ∈ R D ), FL constrains the clusters to be points C ⊆ F in a set of candidate locations F. Hence, we obtain a link between combinatorial Bayesian models and FL allowing us to apply algorithms with known approximation bounds to Bayesian inspired nonparametric models. As we will see in Section 3, our OCC algorithm provides constant-factor approximations for both FL and DP-means. Facility location has been studied intensely. We build on the online facility location (OFL) algorithm described by Meyerson [17]. The OFL algorithm processes each data point x serially in a single pass by either adding x to the set of clusters with probability min(1, min µ∈C x − µ 2 /λ 2 ) or assigning x to the nearest existing cluster. Using OCC we are able to construct a distributed OFL algorithm (Alg. 4) which is nearly identical to the OCC DP-Means algorithm (Alg. 3) but which provides strong approximation bounds. The OCC OFL algorithm differs only in that clusters are introduced and validated stochasticallythe validation process ensures that the new clusters are accepted with probability equal to the serial algorithm. BP-means is an algorithm for learning collections of latent binary features, providing a way to define groupings of data points that need not be mutually 6 Algorithm 4: Parallel OFL Input: Same as DP-Means Output: Accepted cluster centers C Input: Set of proposed cluster centerŝ Output: Accepted cluster centers C Figure 2 : The OCC algorithm for Online Facility Location (OFL). exclusive or exhaustive like clusters. As with serial DP-means, there are two phases in serial BP-means (Alg. 7). In the first phase, each data point x i is labeled with binary assignments from a collection of features (z ik = 0 if x i doesn't belong to feature k; otherwise z ik = 1) to construct a representation x i ≈ k z ik f k . In the second phase, parameter values (the feature means f k ∈Ĉ) are updated based on the assignments. The first step also includes the possibility of introducing an additional feature. While the second phase is trivially parallel, the inherently serial nature of the first phase combined with the infrequent introduction of new features points to the usefulness of OCC in this domain. The OCC parallelization for BP-means follows the same basic structure as OCC DP-means. Each transaction operates on a data point x i in two phases. In the first, analysis phase, the optimal representation k z ik f k is found. If x i is not well represented (i.e., x i − k z ik f k > λ), the difference is proposed as a new feature in the second validation phase. At the end of epoch t, the proposed features {f Finally, to update the feature means, let F be the K-row matrix of feature means. The feature means update F ← (Z T Z) −1 Z T X can be evaluated as a single transaction by computing the sums in parallel. We present the pseudocode for the OCC parallelization of BP-means in Appendix A. We now establish the correctness and scalability of the proposed OCC algorithms. In contrast to the coordination-free pattern in which scalability is trivial and correctness often requires strong assumptions or holds only in expectation, the OCC pattern leads to simple proofs of correctness and challenging scalability analysis. However, in many cases it is preferable to have algorithms that are correct and probably fast rather than fast and possibly correct. We first establish serializability: Theorem 3.1 (Serializability). The distributed DP-means, OFL, and BP-means algorithms are serially equivalent to DP-means, OFL and BP-means, respectively. The proof (Appendix B) of Theorem 3.1 is relatively straightforward and is obtained by constructing a permutation function that describes an equivalent serial execution for each distributed execution. The proof can easily be extended to many other machine learning algorithms. Serializability allows us to easily extend important theoretical properties of the serial algorithm to the distributed setting. For example, by invoking serializability, we can establish the following result for the OCC version of the online facility location (OFL) algorithm: Lemma 3.2. If the data is randomly ordered, then the OCC OFL algorithm provides a constant-factor approximation for the DP-means objective. If the data is adversarially ordered, then OCC OFL provides a log-factor approximation to the DP-means objective. The proof (Appendix B) of Lemma 3.2 is first derived in the serial setting then extended to the distributed setting through serializability. In contrast to divide-and-conquer schemes, whose approximation bounds commonly depend multiplicatively on the number of levels [18] , Lemma 3.2 is unaffected by distributed processing and has no communication or coarsening tradeoffs. Furthermore, to retain the same factors as a batch algorithm on the full data, divide-and-conquer schemes need a large number of preliminary centers at lower levels [18, 2] . In that case, the communication cost can be high, since all proposed clusters are sent at the same time, as opposed to the OCC approach. We address the communication overhead (the number of rejections) for our scheme next. Scalability The scalability of the OCC algorithms depends on the number of transactions that are rejected during validation (i.e., the rejection rate). While a general scalability analysis can be challenging, it is often possible to gain some insight into the asymptotic dependencies by making simplifying assumptions. In contrast to the coordination-free approach, we can still safely apply OCC algorithms in the absence of a scalability analysis or when simplifying assumptions do not hold. To illustrate the techniques employed in OCC scalability analysis we study the DP-Means algorithm. The scalability limiting factor of the DP-Means algorithm 8 is determined by the number of points that must be serially validated. In the following theorem we show that the communication cost only depends on the number of clusters and processing resources and does not directly depend on the number of data points. The proof is in App. C. Theorem 3.3 (DP-Means Scalability). Assume N data points are generated iid to form a random number (K N ) of well-spaced clusters of diameter λ: λ is an upper bound on the distances within clusters and a lower bound on the distance between clusters. Then the expected number of serially validated points is bounded above by P b + E [K N ] for P processors and b points per epoch. Under the separation assumptions of the theorem, the number of clusters present in N data points, K N , is exactly equal to the number of clusters found by DP-Means in N data points; call this latter quantity k N . The experimental results in Figure 3 suggest that the bound of P b + k N may hold more generally beyond the assumptions above. Since the master must process at least k N points, the overhead caused by rejections is P b and independent of N . To analyze the total running time, we note that after each of the N/(P b) epochs the master and workers must communicate. Each worker must process N/P data points, and the master sees at most k N + P b points. Thus, the total expected running time is O(N/(P b) + N/P + P b). For our experiments, we generated synthetic data for clustering (DP-means and OFL) and feature modeling (BP-means). The cluster and feature proportions were generated nonparametrically as described below. All data points were generated in R 16 space. The threshold parameter λ was fixed at 1. Clustering: The cluster proportions and indicators were generated simultaneously using the stick-breaking procedure for Dirichlet processes-'sticks' are 'broken' on-the-fly to generate new clusters as and when necessary. 1 For our experiments, we used a fixed concentration parameter θ = 1. Cluster means were sampled µ k ∼ N (0, I 16 ), and data points were generated at x i ∼ N (µ zi , 1 4 I 16 ). Feature modeling: We use the stick-breaking procedure of [20] to generate feature weights. Unlike with Dirichlet processes, we are unable to perform stickbreaking on-the-fly with Beta processes. Instead, we generate enough features so that with high probability (> 0.9999) the remaining non-generated features will have negligible weights (< 0.0001). The concentration parameter was also fixed at θ = 1. We generated feature means f k ∼ N (0, I 16 ) and data points  To test the efficiency of our algorithms, we simulated the first iteration (one complete pass over all the data, where most clusters / features are created and thus greatest coordination is needed) of each algorithm in MATLAB. The number of data points, N , was varied from 256 to 2560 in intervals of 256. We also varied P b, the number of data points processed in one epoch, from 16 to 256 in powers of 2. For each value of N and P b, we empirically measured k N , the number of accepted clusters / features, and M N , the number of proposed clusters / features. This was repeated 400 times to obtain the empirical averagê , the number of rejections. For OCC DP-means, we observeÊ[M N − k N ] is bounded above by P b (Fig. 3a) , and that this bound is independent of the data set size, even when the assumptions of Thm 3.3 are violated. (We also verified that similar empirical results are obtained when the assumptions are not violated; see Appendix C.) As shown in Fig. 3b and Fig. 3c the same behavior is observed for the OCC OFL and OCC BP-means algorithms. We also implemented the distributed algorithms in Spark [25] , an open-source cluster computing system. The DP-means and BP-means algorithms were bootstrapped by pre-processing a small number of data points (1/16 of the first P b points)-this reduces the number of data points sent to the master on the first epoch, while still preserving serializability of the algorithms. Our Spark implementations were tested on Amazon EC2 by processing a fixed data set on 1, 2, 4, 8 m2.4xlarge (memory-optimized, quadruple extra large, with 8 virtual cores and 64.8GiB memory) instances. 1 We chose to use stick-breaking procedures because the Chinese restaurant and Indian buffet processes are inherently sequential. Stick-breaking procedures can be distributed by either truncation, or using OCC! is divided by that using 1 machine (P = 8). Ideally, the runtime with 2, 4, 8 machines (P = 16, 32, 64) should be respectively 1/2, 1/4, 1/8 of the runtime using 1 machine. OCC DP-means and BP-means obtain nearly perfect scaling for all iterations. OCC OFL rejects a lot initially, but quickly gets better in later epochs. Ideally, to process the same amount of data, an algorithm and implementation with perfect scaling would take half the runtime on 8 machines as it would on 4, and so on. The plots in Figure 4 shows this comparison by dividing all runtimes by the runtime on one machine. DP-means: We ran the distributed DP-means algorithm on 2 27 ≈ 134M data points, using λ = 2. The block size b was chosen to keep P b = 2 23 ≈ 8M constant. The algorithm was run for 5 iterations (complete pass over all data in 16 epochs). We were able to get perfect scaling (Figure 4a ) in all but the first iteration, when the master has to perform the most synchronization of proposed centers. OFL: The distributed OFL algorithm was run on 2 20 ≈ 1M data points, using λ = 2. Unlike DP-means and BP-means, we did not perform bootstrapping. Also, OFL is a single pass (one iteration) algorithm. The block size b was chosen such that P b = 2 16 ≈ 66K data points are processed each epoch, which gives us 16 epochs. Figure 4b shows that we get no scaling in the first epoch, where all the work is performed by the master processing all P b data points. In later epochs, the master's workload decreases as fewer data points are proposed, but the workers' workload increases as the total number of centers increases. Thus, scaling improves in the later epochs. BP-means: Distributed BP-means was run on 2 23 ≈ 8M data points, with λ = 1; block size was chosen such that P b = 2 19 ≈ 0.5M is constant. Five iterations were run, with 16 epochs per iteration. As with DP-means, we were able to achieve nearly perfect scaling; see Figure 4c . Others have proposed alternatives to mutual exclusion and coordination-free parallelism for machine learning algorithm design. Newman [19] proposed trans-11 forming the underlying model to expose additional parallelism while preserving the marginal posterior. However, such constructions can be challenging or infeasible and many hinder mixing or convergence. Likewise, Lovell [15] proposed a reparameterization of the underlying model to expose additional parallelism through conditional independence. Additional work similar in spirit to ours using OCC-like techniques includes Doshi-Velez et al. [9] who proposed an approximate parallel sampling algorithm for the IBP which is made exact by introducing an additional MetropolisHastings step, and Xu and Ihler [24] who proposed a look-ahead strategy in which future samples are computed optimistically based on the likely outcomes of current samples. A great amount of work addresses scalable clustering algorithms [8, 7, 10] . Many algorithms with provable approximation factors are streaming algorithms [18, 22, 6] and inherently use hierarchies, or related divide-and-conquer approaches [2] . The approximation factors in such algorithms multiply across levels [18] , and demand a careful tradeoff between communication and approximation quality that is obviated in our framework. Other approaches use core sets [5, 11] . A lot of methods [2, 3, 22] first collect a set of centers and then re-cluster them, and therefore need to communicate all intermediate centers. Our approach avoids that, since a center causes no rejections in the epochs after it is established: the rejection rate does not grow with K. Still, as our examples demonstrate, our OCC framework can easily integrate and exploit many of the ideas in the cited works. In this paper we have shown how optimistic concurrency control can be usefully employed in the design of distributed machine learning algorithms. As opposed to previous approaches, this preserves correctness, in most cases at a small cost. We established the equivalence of our distributed OCC DP-means, OFL and BP-means algorithms to their serial counterparts, thus preserving their theoretical properties. In particular, the strong approximation guarantees of serial OFL translate immediately to the distributed algorithm. Our theoretical analysis ensures OCC DP-means achieves high parallelism without sacrificing correctness. We implemented and evaluated all three OCC algorithms on a distributed computing platform and demonstrate strong scalability in practice. We believe that there is much more to do in this vein. Indeed, machine learning algorithms have many properties that distinguish them from classical database operations and may allow going beyond the classic formulation of optimistic concurrency control. In particular we may be able to partially or probabilistically accept non-serializable operations in a way that preserves underlying algorithm invariants. Laws of large numbers and concentration theorems may provide tools for designing such operations. Moreover, the conflict detection mechanism can be treated as a control knob, allowing us to softly switch between stable, theoretically sound algorithms and potentially faster coordination-free 12 algorithms.
Recently, proposals have emerged to leverage optimistic concurrency control (OCC) from database serializability theory for function synchronization in the context of machine learning problems such as unsupervised clustering REF .
Optimistic Concurrency Control for Distributed Unsupervised Learning
ArXiv
ArXiv
Computer Science Mathematics