分布式机器学习之SSP模型

More Effective Distributed ML via a Stale Synchronous Parallel Parameter Server
摘要
引言
SSP模型的计算过程
SSPtable：一个高效的SSP系统
SSP理论分析
参考文献

分布式机器学习之SSP模型论文解析。

More Effective Distributed ML via a Stale Synchronous Parallel Parameter Server

摘要

本文提出一个分布式机器学习的参数服务器系统，该系统遵循SSP计算模型。模型最大化机器学习算法的有用工作时间并提供收敛保证。参数服务器提供了一个易于使用的共享接口，用于对机器学习模型的值(参数和变量)进行读写访问，并且SSP允许分布式Worker从本地缓存中读取旧的、过时的值，而不是从中央存储中获取它们。这大大增加了Worker计算时间的比例而不是等待。此外，SSP通过限制延迟阈值的最大值确保算法正确性。本文给出SSP的准确性证明并通过实验结果证明SSP在不同的机器学习问题上相比完全同步和异步策略具有更快的算法收敛速度。

引言

现代应用在等待下一代机器智能系统上已经提出了前所未有的可扩展性挑战。这些可扩展性需求至少产生于两方面：1)海量数据，例如多达数一节点的社交图[10, 25]；2)巨大的模型，例如包含数十亿参数的Google大脑深层神经网络[9]。尽管存在支持简化方法的手段和理论，例如采样数据或使用小模型，但是这种方式不能很好地适应需求，迫切需要健全和有效的分布式机器学习方法。近年来，对分布式机器学习在两个方向上取得了重大进展：1)利用现有通用且简单的分布式系统实现有限选择的机器学习模型的并行版本，在诸如循环延迟[17, 1]、模型预分区[12]、无锁更新[21]等并行化方案下，这些模型可被证明具有强理论保证，批量同步并行[5]，甚至是没有同步[28]-这些方案易于实现，但是可能没有完全利用分布式集群的计算能力。2)构建高吞吐分布式机器学习体系结构或算法实现，这些体系结构或算法实现以显著地系统贡献为特征，但是理论分析相对较少，例如GraphLab[18]、Spark[27]、Pregel[19]、YahooLDA[2]。

虽然上述工作本身具有重要的贡献，但是分布式机器学习追求如下系统：1)可以最大限度地释放任意给定大小的集群中的组合计算能力(通过花费更多的时间用于有效的计算，更少的时间用于通信等待)，2)支持机器学习方法的推理，3)具有正确性保证。本文中，我们使用参数服务器[22, 2]实现，将其定义为贡献共享键值对存储用于读取/更新同步模型，提供中心化存储模型(可以以分布式方式实现)。键值对存储为Workers提供了易于编程的读写访问，并且同步模型是Worker最大化计算时间(相对于与服务器通信)，同时提供算法正确性保证。

为此，我们提出一个使用延迟同步并行(SSP)计算模型的参数服务器，用于分布式机器学习算法，这些算法被并行化分布在多台机器上实现。在SSP中，Workers可以产生参数Θ的更新δ，这里更新遵循关联的、可交换的形式：Θ<-Θ+δ。因此，当前的Θ是所有Worker产生的更新δ。当一个Worker需要Θ，SSP模型将会给它一个旧版本的(或延迟的)不包括当前更新δ的Θ。更形式地说，一个Worker在迭代步c读取Θ将会看到所有从迭代步0到迭代步c-s-1所产生的δ，s>=0是用户控制的延迟阈值。另外，Worker可能会看到一些超过迭代步的c-s-1更新。SSP系统的期望是提供尽可能多的更新，不丢失任何比给定时间更早的更新——一个称谓有界延迟的概念[24]。这种方法的实际效果是双重的：1)Worker可以提供更多的计算而不是等待其他Worker；2)Worker花费更少时间和参数服务器通信，更多的时间用于计算。有界延迟区分了SSP和循环延迟系统17, 1，BSP系统诸如Hadoop(Worker必须在每一轮等待彼此结束)，或完全异步系统2。

我们实现的SSP参数服务器使用了基于表的接口，称之为SSPtable，它为许多模型和应用提供了宽泛的分布式机器学习算法。SSPtable本身也可以以分布式的方式运行，以便于提高性能或支持参数数量太大而不能再一台机器上运行的应用程序。此外，SSPtable利用了有界延迟从而最大化机器学习算法的性能，通过读取参数Θ在Worker上的缓存，仅当SSP模型需要时从参数服务器读取参数。因此Worker(1)花费更少时间用于等待彼此；(2)花费更少的时间用于和参数服务器通信。此外，SSPtable帮助慢的、落后的Worker赶上来，提供一个基于系统的解决方案来解决类似last reducer在Hadoop上的问题(同时我们注意到基于理论解决方案也是可行的)。SSPtable可以在多服务机器上运行(称之为shards),因此在集群上划分其工作量；以这种方式，SSPtable可以(4)同时为多台Worker服务，并(5)支持单机无法运行的非常大的模型。最后，SSPtable的server程序也可以运行在Worker上，(6)提供一个简单但高效的策略用于分配在Worker和参数服务器间分配节点。

我们的理论分析表明(1)SSP推广了BSP并(2)随机梯度下降算法在SSP不仅仅收敛，此外收敛速度至少和循环延迟系统[17, 1]一样快(依赖实现可能会更快)。此外，我们实现了SSP，SSPtable，支持各种不同的算法或模型，并且我们在许多不同的例子下证明：(a)使用矩阵分解的随机梯度下降[12]，(b)Topic Modeling with collapsed Gibbssampling[2]，(c)使用L1回归的并行坐标下降[5]。我们的实验结果表明，对于这三个模型和算法，(1)SSP比BSP有更快的收敛速度(快几倍)，(2)SSP比ASP有更快的收敛速度。我们解释了SSPtable在每次迭代的算法进度（质量）和单位时间（数量）执行方面有更好的性能，并表明SSPtable在质量和数量之间达到了BSP和全异步系统所错过的“最佳点”。

SSP模型的计算过程

我们开始正式解释SSP：假定有P个Worker，每个Worker每隔一段时间都对共享参数进行附加更新x<-x+u，这段时间称为时钟。时钟类似于迭代步，并且表示机器学习算法运行过程中的一小段。每个Worker有自己的整数值时钟c，并且Worker仅仅在每个时钟结束时提交自己的更新。更新对于其他Worker可能不会立即可见。换言之，Worker仅能看到来自全部更新的一部分(旧的、过时的)。通过使用旧数据，Worker能从本地缓存中快速取回更新而不是通过较慢的网络去参数服务器查询。给定一个用户定义的延迟阈值s>=0，SSP执行下面的边界延迟条件：

由于最慢和最快的Worker时钟差必须<=s，一个Worker在时钟c读取x时将会看到时间戳[0, c-s-1]的更新以及在[c-s, c+s-1]这段子集的自适应更新。注意到当s=0时，[0, c-1]这段更新将完全可取，而自适应更新为空，SSP退化成BSP。

SSPtable：一个高效的SSP系统

理想的SSP实现将充分利用SSP的边界延迟属性赋予的余地，以平衡Worker读取时间和共享数据的新鲜度。本节描述SSPtable的初始实现，它是一个复合SSP模型的参数服务器，可以在多个服务器(分布式)上运行。我们在SSPtable上的实验表明，SSP确实可以提高几种机器学习模型和算法的收敛速度，此外进一步优化缓存管理策略可以进一步提高SSPtable的性能。

SSPtable采用了客户端-服务器架构。客户端使用客户端库访问共享参数，客户端库维持单机范围内的线程缓存以及每个线程可选的线程缓存。后者有利于提高性能，因为当一个客户端的机器学习程序在多核心中每个上执行多个工作线程时，减少了线程间同步。参数被划分到多台机器上，正常配置将包括每个客户端机器上的服务器进程。SSPtable编程遵循一个简单的基于表的API来读取/写入共享参数X：

任意数量的read_row()和inc()可以在clock()之间被多次调用。不同的Worker线程可以处于不同的时钟周期，但是有界延迟要求最快线程与最慢线程之间的时钟差不能高于s个时钟。当超出s个时钟时，SSPtable要求最快线程调用read_row()时阻塞或等待，直到最慢进程赶上来。为了维持read-my-writes属性，我们使用write-back策略：所有的写入立即提交到线程缓存，并在调用clock()时更新进程缓存和服务器缓存。

为了维持有界延迟并调用read_row()时最小化等待时间，SSPtable使用如下缓存策略：让每个table-row在线程或进程缓存中被赋予各自的时钟戳r_thread和r_proc。让每个线程Worker被赋予一个时钟c，等价于它调用clock()的次数。当一个线程在时钟c时需要一个table-row,首先检查线程缓存。如果row的缓存时钟戳r_thread>=c-s，那么它读取这个row。否则，它检查进程缓存。如果row的缓存时钟戳r_proc>=c-s，那么它读取这个row。此时，并没有发起网络连接。然而，如果两次缓存均未命中，接下来将会向参数服务器发起网络请求(强制线程等待回复)。服务器返回table-row的视图以及时钟c_server。因为最快的线程和最慢的线程时钟差不超过s，并且因为无论何时调用线程的更新被发送到服务器，返回的服务器视图总会满足有界延迟的需求。从服务器取回row后，进程或线程缓存中相关的entry以及r_thread和r_proc将被来自服务器视图和c_server覆盖。

缓存协议一个有益的结果是最慢的线程仅在每s个clock执行昂贵的服务器读取。更快的线程频繁地执行服务器读取，如果它们一直等待最慢线程的更新，那么它们可以在每个时钟频繁地执行服务器读取。BSP的每个线程在每个时钟上必须从服务器读取。因此，SSP不仅减少了总的网络流量(因此减少了所有服务器读取的等待时间)，而且允许较慢的线程避免在某些迭代中执行服务器读取。这样，慢节点自然地能赶上来，反过来，允许快进程继续执行而不是等待慢进程。此方法中，SSP最大化每台机器用于计算的时间，而不是等待。

SSP理论分析

我们假定P个Worker每个一段时间(时钟周期)写更新。令u_p,c为p号Worker在时钟c的产生的更新，那么写操作为：x<-x+u_p,c。更新u_p,c是系统状态x的函数，在SSP模型下，相比于真正的x不同的Worker将会看到不同的，具有噪音的状态。令x_p,c是p号Worker在时钟c读到的噪音状态，推导出对于某些函数G有：u_p,c=G(x_p,c)。我们现在正式地重新声明边界延迟，这是SSP的关键条件，限制x_p,c可能的值可以采取:

参考文献

[1] A. Agarwal and J. C. Duchi. Distributed delayed stochastic optimization. In Decision and Control (CDC),
2012 IEEE 51st Annual Conference on, pages 5451–5452. IEEE, 2012.
[2] A. Ahmed, M. Aly, J. Gonzalez, S. Narayanamurthy, and A. J. Smola. Scalable inference in latent variable
models. In WSDM, pages 123–132, 2012.
[3] N. R. Alexandros Labrinidis. Balancing performance and data freshness in web database servers. pages
pp. 393 – 404, September 2003.
[4] M. Bouzeghoub. A framework for analysis of data freshness. In Proceedings of the 2004 international
workshop on Information quality in information systems, IQIS ’04, pages 59–67, 2004.
[5] J. K. Bradley, A. Kyrola, D. Bickson, and C. Guestrin. Parallel coordinate descent for l1-regularized loss
minimization. In International Conference on Machine Learning (ICML 2011), June 2011.
[6] L. Bright and L. Raschid. Using latency-recency profiles for data delivery on the web. In Proceedings of
the 28th international conference on Very Large Data Bases, VLDB ’02, pages 550–561, 2002.
[7] J. Cipar, G. Ganger, K. Keeton, C. B. Morrey, III, C. A. Soules, and A. Veitch. LazyBase: trading
freshness for performance in a scalable database. In Proceedings of the 7th ACM european conference on
Computer Systems, pages 169–182, 2012.
[8] J. Cipar, Q. Ho, J. K. Kim, S. Lee, G. R. Ganger, G. Gibson, K. Keeton, and E. Xing. Solving the straggler
problem with bounded staleness. In HotOS ’13. Usenix, 2013.
[9] J. Dean, G. Corrado, R. Monga, K. Chen, M. Devin, Q. Le, M. Mao, M. Ranzato, A. Senior, P. Tucker,
K. Yang, and A. Ng. Large scale distributed deep networks. In NIPS 2012, 2012.
[10] Facebook. www.facebook.com/note.php?note_id=10150388519243859, January 2013.
[11] C. J. Fidge. Timestamps in Message-Passing Systems that Preserve the Partial Ordering. In 11th Australian
Computer Science Conference, pages 55–66, University of Queensland, Australia, 1988.
[12] R. Gemulla, E. Nijkamp, P. J. Haas, and Y. Sismanis. Large-scale matrix factorization with distributed
stochastic gradient descent. In KDD, pages 69–77. ACM, 2011.
[13] L. Golab and T. Johnson. Consistency in a stream warehouse. In CIDR 2011, pages 114–122.
[14] C.-T. Huang. Loft: Low-overhead freshness transmission in sensor networks. In SUTC 2008, pages
241–248, Washington, DC, USA, 2008. IEEE Computer Society.
[15] L. Lamport. Time, clocks, and the ordering of events in a distributed system. Commun. ACM, 21(7):558–
565, July 1978.
[16] J. Langford, L. Li, and A. Strehl. Vowpal wabbit online learning project, 2007.
[17] J. Langford, A. J. Smola, and M. Zinkevich. Slow learners are fast. In Advances in Neural Information
Processing Systems, pages 2331–2339, 2009.
[18] Y. Low, G. Joseph, K. Aapo, D. Bickson, C. Guestrin, and M. Hellerstein, Joseph. Distributed GraphLab:
A framework for machine learning and data mining in the cloud. PVLDB, 2012.
[19] G. Malewicz, M. H. Austern, A. J. Bik, J. C. Dehnert, I. Horn, N. Leiser, and G. Czajkowski. Pregel: a
system for large-scale graph processing. In Proceedings of the 2010 International Conference on Management
of Data, pages 135–146. ACM, 2010.
[20] F. Mattern. Virtual time and global states of distributed systems. In C. M. et al., editor, Proc. Workshop
on Parallel and Distributed Algorithms, pages 215–226, North-Holland / Elsevier, 1989.
[21] F. Niu, B. Recht, C. Re, and S. J. Wright. Hogwild!: A lock-free approach to parallelizing stochastic ´
gradient descent. In NIPS, 2011.
[22] R. Power and J. Li. Piccolo: building fast, distributed programs with partitioned tables. In Proceedings
of the USENIX conference on Operating systems design and implementation (OSDI), pages 1–14, 2010.
[23] U. Rohm, K. B ¨ ohm, H.-J. Schek, and H. Schuldt. Fas: a freshness-sensitive coordination middleware for ¨
a cluster of olap components. In VLDB 2002, pages 754–765. VLDB Endowment, 2002.
[24] D. Terry. Replicated data consistency explained through baseball. Technical Report MSR-TR-2011-137,
Microsoft Research, October 2011.
[25] Yahoo! http://webscope.sandbox.yahoo.com/catalog.php?datatype=g, 2013.
[26] H. Yu and A. Vahdat. Design and evaluation of a conit-based continuous consistency model for replicated
services. ACM Transactions on Computer Systems, 20(3):239–282, Aug. 2002.
[27] M. Zaharia, M. Chowdhury, M. J. Franklin, S. Shenker, and I. Stoica. Spark: cluster computing with
working sets. In Proceedings of the 2nd USENIX conference on Hot topics in cloud computing, 2010.
[28] M. Zinkevich, M. Weimer, A. Smola, and L. Li. Parallelized stochastic gradient descent. Advances in
Neural Information Processing Systems, 23(23):1–9, 2010.