mxnet分布式2

ps-lite论文阅读 https://www.usenix.org/system/files/conference/osdi14/osdi14-paper-li_mu.pdf

1 Intruduction

数据很大，分布式已经是趋势，节点之间共享参数是必须，但是遇到三个挑战

1. 高带宽的获取共享参数
2. 很多机器学习是顺序的，同步和高延迟的阻碍影响了性能
3. 高容错率

1.1 ps-lite提供了5个核心功能特征

1. 异步高效通信
2. 灵活的一致性模型，平衡同步与延迟，算法设计者自己平衡算法收敛与效率
3. 节点弹性伸缩
4. 高容错率且耐用，快速从修复中恢复，时钟向量机制使得网络失败或者分离后行为明确
5. 容易使用

1.2 工程实现的挑战

1. 参数在节点之间的高效通信
2. 容错性，一台机器挂了，整个任务还是在run

2机器学习

2.1 目标

机器学习的目标一般是最小化目标函数，最优解被找到或者模型收敛，训练结束，一般需要处理的数据量会很大，在这些大量数据上执行算法是本文的目的

2.2 最小化误差(risk minimization)

机器学习最直观的转化是最小化误差，risk指的是预测值和标准值之间的差值，比如预测股票，预测的股价和后来真是的股票价格之间的差值就是risk

训练数据包含n个样本，xi是第i个样本，且经常是一个长度为d第矢量，n和d可能是十亿和万亿级别的数据，很多场合下，样本xi有一个label yi 和它对应，在广告预测中，yi则1对应点击了，-1对应没有被点击。

基于最小化误差方法能学习一个模型，该模型之后可以用来预测其他新的样本，为了预测未来的一个广告是否会被点击，系统将对'clikness'求和，然后机遇这一坨参数决定未来的广告是否会被点击，就是新输入经过训练好的参数提取特征，得到结果

在很多学习算法中，训练数据和模型大小息息相关，模型越大，或者越详细，之后能做出的预测就越准确，除非训练集太小，模型太大，训练集太小会造成过拟合，模型记住了每一个样本的特征，导致它失去了泛化能力，模型太小则捕捉不到感兴趣的特征

正则化最小风险是找到模型复杂度和训练误差之间的平衡，下面公式第一项是训练误差，第二项是模型复杂度，它的目标是使两者的和加起来最小，前面一项是loss，后面一项是regularizer

image

image

image

在表2和算法1中，数据被分成很多小份到每个节点上去联合计算w，每个worker计算自己所得到的那份梯度，server去聚合所有worker得到的梯度，乘以学习率，然后进行下一轮迭代

最耗时的操作是计算子梯度以便更新梯度，该任务分到每个worker中去做，每个worker执行w�xik，对于太大的w这是不可行的，幸运的是一个worker只需要知道它的训练数据所对应的那一部分参数w就行了

比如在广告点击的例子中，如果很少的广告包括OSDI2014这个广告语，那么大部分的worker就不需要更新去更新这个广告词所对应的那一部分参数w，由于总的参数无法在一台机器上放得下，每一台机器需要的那一部分参数可以存放在本地，为了证明这个结论，我们随机的把数据分配到不同的worker，然后计算每个worker基于那部分数据所需要的参数的平均大小，具体的细节在5.1章节中。图3显示，100个worker的集群，每个worker只需要参数的7.8%，1000个只要0.15%

image

2.3 生成模型

在第二种机器学习算法中，一个样本对应的label是未知的，这类学习叫无监督学习，它们尝试捕捉潜在的数据结构，一个常见的例子是主题模型：给一堆文档，推导出每个文档的主题

举个例子，当跑sosp2013的proceeding这个程序的时候，一个算法可能会产生主题‘distribute system’, 'machine learnling', 'performance',算法通过论文内容产生这些主题，而不是通过外部给的一个主题列表，在一些场合比如个性化推荐，问题的规模会变得很大，亟待用分布式并行处理

由于数据的规模，这些算法只有应用在第一代参数服务器上才有商业价值，用文档怎么产生当前的主题估计的参数必须共享，这是一个关键的挑战

一个流行的主题建模是LDA，这个统计模型语其他的相当不同，模型语算法1比较相似，不同的是计算的不再是梯度吗，而是文档多大程度能被解释,对于每一个文档，这个算法需要额外的数据当每次文档被用到的时候，因为每次文档被处理的时候，文档与元数据都会被存入取出。正如前面章节所描述，每个worker存储了它所处理的文档的关键字，因而，采用分布式系统能处理更大的模型

3架构

一个ps实例可以跑多于一个算法模型，ps由一个ps node group组成，worker由workergroup组成，一个server node维护着一份共享参数，所有的server node之间可以相互通信，所有的server node共享一个server manage,server manager维护参数的一致性，比如收集节点的心跳信息，参数的分配等等

image

每一个worker group运行一个程序，一个worker保存部分数据并对它进行计算，比如深度学习中的梯度计算，worker只与serveer node之间通信，而不会在workers之间它们自己通信，和server通信比如将计算好的梯度推送到server或者从server获取聚合后到参数，每一个worker group都有一个worker scheduler，它负责向每个worker分配任务，并且管理它们的生命周期等等

参数服务器支持独立的参数命名空间，这使得不同worker group之间的参数集互相独立，同时不同的worker group也可以使用一个相同的命名空间，这样可以以更大的并行程度去解决一个深度学习问题，另外一个例子是模型经常被节点访问，比如在线服务来访问这个模型(这坨参数)，同时当新的参数到一个worker时并计算出结果时，模型被这个worker所更新

参数服务器被设计用来简化分布式的应用，如第二章所提到的那些应用。被共享的参数用k-v对来表示，这样对代数运算更容易被处理，详细介绍在3.2节，这些参数被分布式的存储在server group不同的节点中，任何节点可以从服务器pull参数和push本地的参数(梯度)到server node。在默认情况下，任务是由worker来完成了，少数情况下任务也可以由server节点完成。通过任务状态依赖图和与哪一部分参数通信，ps让算法开发者可以灵活的选择一致模型。

3.1 (key, value)向量

被不同的worker node共享的模型可以用key-value的键值对所表示，比如在最小化损失函数的例子中，key是特征ID,values是它所对应的权重，对于LDA,key-value对是word-ID-topicID,模型的每一个实体可以被本地或远程的读取或写入，键值对的概念被很多框架所采用。

ps框架在也采取了key-value这种策略，并且给予这样的观点：典型的机器学习算法把模型当作线性代数对象来对待，在目标函数和最小化风险的例子中，w都被当作向量对待。把这些键值对当作线性代数对象，parameter server可以应用同样概念的运算规则，比如向量加减乘除等，将向量的代数运算也移植到key-value这样的对象上来。

为了支持这些优化，ps框架假设这些key是按顺序排列的，赋予这些键值对矩阵的语义，key不能为0，这把大量的编程问题简化成了实现优化算法，不仅仅是为了更高效的code，key-value这套接口还借力了CPU的一些线性代数多线程编程库比如BLAS,PLACK,ATLAS等等，简单的理解就是模型用键值对表示，这样有一堆好处。

3.2 Range Push and Pull

Range push就是将指定范围的key-value推送，公式：w.push(R, dest),如果R只是一个值，那就是push单个key对应的value，如果R对应的范围是所有参数的key，那就是一次完整的push，即将全部的参数推送到服务器。这个接口也可以扩展到和参数w共享key到其他数据通信，比如对于算法1中从worker中向server更新梯度，可以更新带范围的梯度，这样写w.push(R, g, dest)

3.3用户自定义的函数-server node上执行

除了聚合worker的参数到server上，server node上还可以执行用户自定义的参数，因为server上有更加完整，更实时的参数，算法1将各个worker的梯度在server上进行聚合，在算法3中有一些复杂的运算，则需要在server上进行，在上下的处理上，几乎所有的操作都在server端

3.4异步任务和依赖

一个任务被远程调用者所触发，它可以是worker node向server node发起的push／pull请求，也可以是一个自定义的函数scheduler发送到任何node上，一个task可能包含许多子任务，比如在算法1中一个数据迭代(workeriter)包含一个pull和一个push

任务被异步执行，调用者发出任务请求后就自己干别动事情去了，调用方表记一个任务为执行完当且仅当它收到任务的返回，比如用户定义的函数被返回，pull或push的参数被pull／push成功的返回码被接受到，任务处理者标记一个任务被处理的的标志是这个任务已经完成且所有的子任务也被执行完毕

默认情况下任务是并行执行的，在图5中，iter10和iter11是并行执行的，但是iter12的计算却是依赖于iter11的结果的，iter12要等到iter11的计算结果push完后它才能开始

image

任务依赖可以帮助算法实现，比如在算法1中，只有所有worker的梯度被push到server的时候，server node才开始做参数聚合，依赖的第二个作用是用于实现灵活的一致性

3.5灵活的一致性

通过并行地使用CPU，磁盘，带宽等资源，可以提升系统等性能，但是这会使得数据出现不一致的现象。数据不一致的举例：在图5中，iter10和iter11是并行的，因而10和11获取的是一样的参数，所以它们得到的数据是不一致的，但是12和11就是一致的，因为12依赖了11，10和11计算出来的东西一样，这个不一致导致了收敛变慢，如果用这个优化算法来计算算法1，那么就会出现上面所说的收敛变慢，但是在有一些例子中，算法对数据不一致性不是那么的敏感，每次迭代只有一部分的参数被更新，比如在算法3中，并行的是特征(这个并行暂时也不是特别明白，模型并行),最好的折中往往要取决于各种参数，比如模型对数据不一致性的敏感程度，特征对数据的相关性，硬件组件等容量能力等。ps没有规定死用户必须采用哪些方案来适配具体固定的问题，而是算法设计者可以灵活的，根据具体情况选择不同的折衷方案。这里示出3种使用依赖可以实现的模型，它们的有向无环图见图6

image

• Sequential, 一个一个的执行，后者只有在前者执行完才能开始
• Eventual, 各干个的，互不关联
• Bounded delay, 某个任务开始执行还是阻塞取决于在它之间t时刻起的任务已经被执行完毕，如果t设置为0，那么他就是第一种顺序执行，如果是t为无穷，那么它就是第二种，互不影响

以上举例的图是可以动态变化的，比如scheduler想让收敛点快一点，活着计算慢一点，活着有新的节点加入了计算图，那么可以修改这些因素

3.6用户定义的过滤器

基于scheduler，ps可以实现细粒度的控制同步，只有符合过滤器条件的worker的参数才会同步到server上，这样做是因为优化器本身有更多的参数的信息，举个例子,significantly modified filter,在worker上计算出来的条目(entry)-比如梯度，只有超过阈值的才会被推送到server服务器上，在5.1节中，讨论了另外一种KKT的方案，它利用了优化问题的最优条件-只有能影响权重的梯度才会被发送到srver上(这和上文提到的不是一样吗，稀疏)

image

4实现

服务器使用连续的hash(键值对)存储参数见4.3，为了容错，参数会被链式备份，见4.4节；与以往系统不同，ps-lite用‘’基于范围通信‘’优化了服务器，优化了数据和向量时钟

4.1时间向量

由于计算图的复杂性及快速恢复参数的需求，每一个参数都有一个时间戳和它对应，时间戳对于追踪聚合状态或拒绝发送数据有很大的好处，假如有n个node和m个参数，时钟向量需要O(mn)的空间复杂度，这样是不可行且占带宽的

但是ps使用了range-push/pull的策略，那么那一个range的key/value对的时间戳肯定是一样的，这一个range的参数所对应的时间戳可以压缩到一个值，一个参数向量假如包含了k个range set，那么时钟向量只有k个值，对于整个ps系统，只需要O(mk)空间复杂度的时钟向量，m是server节点的个数

4.2消息

一个节点可能发送消息到其他的节点或者节点组，一个消息实体包括一个key-value的列表和一个时钟向量列表，通信与任务的格式都基于这种格式，对于后文碰到的格式，可能都是基于这种形式

消息可能是一个有效列表的子集，在一个range R里面，缺失的key可以分配给一个相同的时间戳，当一个worker向所有的server或一个server group发送消息的时候，或者一个key分配给一个server node改变了的时候，这时候一条消息可能被key range 分开，通过这种办法，我们把数据列表以及它们所对应的时间戳分开(这说明对于一个完整的消息，每个server node只保存了部分参数)

压缩-每次数据迭代所产生的key-value的key其实是一样的，这样可以把key缓存起来，下次push或pull的时候只要push那一坨数据的hash值就行了。values同样可以压缩，worker向服务器更新参数的时候很多值可能都是不变的，举个例子之前介绍的用户自定义的过滤器函数，这个函数只要是0活着阈值以下的都不会被push到服务器端，因而在push的时候肯定push了很多0值到服务器，这样可以对这些0值压缩，ps使用了一个叫做Snappy的库对数据进行压缩

4.3hash一致性

Ps 分key对办法和传统的分布式hash表很像：key和serverID都被插入到图7这样的一个环中，每一个server node管理它的插入点到逆时针方向的下一个server node插入点之间的那些key range,这个节点被称作是这个key range的master，一个物理节点被复制成多分称为虚拟节点，这样做的目的是负载平衡

ps使用一个直接映射DHT的hash一致性算法设计，server manager节点管理ring manager,所有的节点缓存它所对应的key在本地，这样所有的server node 就能够找到它所对应的key range，总之这里的hash一致性的算法与百度一把的hash一致性差不多，就是把server node和key都映射到一个环上，然后数据就近存储到server节点

4.4 Replication and Consistence

每一个server节点存储了k个它的邻居节点的key range，也就是一份key range被k+1个server节点所保存，每个server保存的参数都有其对应参数的key range，每个节点都保存了其他k个server的这个key range表，在上面举例中也就是图7中k=2，server node1保存了server node2和server node3所拥有的key range

master节点数据的任何修改都会被copy到它的slaves机器上，图8展示了work1推送x到server1, server1调用函数f得到新的值，只有这份修改后的数据被拷贝到server2这个操作才算结束。根据图理解起来就是，worker1计算的梯度推送到server1，server1聚合梯度为参数，server1将含有时间戳到参数发送给server2, 这个操作才算结束。

原始的复制会备份k次，每一个worker更新梯度后master server处理了梯度后都会将其复制到备节点的机器上，ps框架提供了新的方法，等到所有的worker的梯度都被server聚合后，再去备份。

image

4.5 server manager

为了增加容错的能力，需要支持添加与删除节点

当添加一个新的节点时：

• 分配key range 给这个新的节点，这可能导致其他节点所维护的key被拆分
• 新节点获取分配给它点数据，并且保存k个备份
• 新节点向其他节点广播信息

分两步获取和range R相关的数据，首先对S节点拷贝所有的key-value对的数据及时钟向量，这可能会导致一个range的时钟向量被拆分，这个过程一样算法2相似，如果这个过程失败，S node会返回之前的状态，这个操作有原子性，第二步，S node不在执行相关的消息请求与处理，同时S node会将所有有关 range R的变化发送给新的节点。一个server节点N如果收到了相关广播，它首先检查自己是否和广播中的R有关系，如果有则作相关的操作

当一个节点离开时：

节点的离开与新增节点类似，通过server node节点的心跳信息，server manager判断一个节点是否还在工作，如果不在了，则将与其相关的range R的数据发送给其他的server node

4.6 worker managerment

增加新的worker和增加新的server node类似但是更简单：

• Taske scheduler 分配新的数据给这个worker
• 新的worker从文件系统中或者已经存在的worker中获取数据，然后从server中pull参数下来进行训练
• Task scheduler广播这一变化，这个变化可能会引起其他数据节点的数据减少

当一个worker离开的时候，算法开发者有两种选择，一个是启动一个新的worker来替代这个挂掉的worker，或者干脆不管这个worker，因为从一坨巨大的数据中恢复一个worker的代价是很大的，另外一个原因是对与分布式训练，失去这个worker对整体的性能其实没什么影响

测评

所有测评基于稀疏的逻辑回归和隐含的狄利克雷分布

所用的两个对比都是基于大的逆天的搜索数据，基于稀疏的逻辑回归的实验主要和其他两个系统A和B做了比较，对比了目标loss下降与所花时间到关系，ps快到不行；另一幅图的结果则展示了ps几乎没有等待时间但是A和B系统缺花了大量的时间阻塞在等待上面，文章中解释了异步，压缩等上文所提到的原因加速了训练的速度，提升资源利用率等