采用参数服务器的大规模分布式机器学习

1. 简介

大规模分布式机器学习三大挑战：

• 访问这些参数（训练数据范围达到1TB到1PB，参数量范围10⁹ ~ 10¹²）需要巨大的网络带宽。
• 很多机器学习算法是有顺序性的。当同步的花费、机器的延迟高时，会导致栅栏效应影响性能。
• 错误容忍是决定性的。学习任务（Learing Tasks）经常在机器可能是不可靠的、工作可能被占用的云环境中执行。

本文描述了第三代参数服务器的开源实现

table2.png

该参数服务器的五个关键特征：

• 高效的通信（Efficient communication）：异步通信模型不阻碍计算，最优化机器学习任务来减少网络负担。
• 灵活的一致性模型（Flexible consistency models）：宽松的一致性进一步隐藏同步花销和延迟。允许算法设计者去平衡算法收敛率与系统效率。
• 灵活的可扩展性（Elastic Scalability）：在无需重启运行中的框架下便可添加新节点。
• 错误容忍和持久（Fault Tolerance and Durability）
• 易用（Ease of Use）

2. 架构

figure2.png

algorithm 1.png

figure4.png

风险最小化问题：学习一个模型能通过future example$x$来预测值$y$，模型由参数$w$组成。
$$F(w)=\sum_{i=1}^nl(x_i+y_i+w)+\Omega(w)$$
loss函数$l(x+y+w)$表示在训练数据中预测错误，正则化函数$\Omega(w)$使模型复杂，因此我们希望$F(w)$最小化。
如上图figure4。参数服务器可以同时跑不止一个算法。
每个服务器节点维持全局共享参数的一部分。为保证可靠性，服务器节点之间可以相互通信来复制、迁移参数。用一个服务器管理节点来维持服务器元数据的一致性，例如节点是否存活，参数划分的分配情况。
每个工作组跑一个应用。工作节点在本地存储一部分训练数据来计算本地梯度等值。工作节点只与服务器节点通信来更新和取回共享参数（工作节点相互之间不通信）。每个工作组设置一个计划表节点用来分配任务给工作节点和监视它们的进程。若增加或移除了工作节点，它将重新安排未完成的任务。

2.1 (Key,Value) Vectors

共享参数以(Key,Value) Vectors的方式传输。

2.2 Range Push and Pull

在节点间数据传输采用Push、Pull操作。

• 在Algorithm 1，每个工作节点Push它全部的本地梯度到参数服务器，然后Pull更新的参数回来。
• 在algorithm 3，每次传输一个范围的keys。

alhorithm3.png

• 参数服务器支持range-based push and pull。 设$R$为一个key的范围，w.push($R$,dest)发送在$R$范围内所有存在的$w$到目的地（可以是一个节点，一个节点群如服务器群）。w.pull($R$,dest)从目的地读取在$R$范围内所有存在的$w$。
• 如果我们把$R$设置为全部key范围，则全部参数向量$w$会被通信，若我们把$R$设置为单个key，则只有独立的entry会被发送。
• 这个接口可以扩展到通信任何分享相同keys作为$w$的本地数据结构。？

2.3 User-Defined Functions on the Server

除了聚集从工作节点上传的数据，服务器节点还可以实行用户定义的函数。

• Algorithm 1，为了更新$w$，服务器节点评估次梯度的正则化矩阵\Omega

2.4 Asynchronous Tasks and Dependency

任务执行时异步的：caller能在发出任务后立即执行下一步的计算。

• 只有在收到callee的回复后caller才将任务标记为完成。
• callee只有在该任务的call已经返回并且所有子任务都完成的情况下才将该任务标记为完成。

figure5.png

如上图，Iterations 10和11是独立的，但12依赖于11。因此，callee在计算Iterations 10的本地梯度后立即开始Iterations 11，而Iterations 12要延迟到pull了11的结果后执行。

2.5 Flexible Consistency

figure6.png

Bounded Delay：设置一个最大延迟时间$\tau$,只有在所有$\tau$前的任务都已经完成后才会开始新的任务。Algorithm 3使用。

2.6 User-defined Filters

参数服务器支持用户定义的过滤器来选择性的同步独立（键，值）对，允许在一个任务对数据一致性进行细粒度的控制。

• significantly modified filter：只push自上次同步后改变的大小超过阈值的entries。
• KKT利用优化问题的最优性条件：worker只push看起来对servers中的权重有影响的梯度

3 实现

1. 服务器使用一致性哈希存储参数（键值对）。（3.3）
2. 对于错误容忍，entires使用链复制。（3.4）
3. 不同于先前的键值对系统，参数服务器充分利用以范围为基础的通信，压缩数据（3.2）和基于向量时钟的范围（3.1）。

3.1 Vector Clock

考虑到复杂任务的依赖图以及快速恢复的需要，每个键值对与一个向量时钟对应，向量时钟记录了在这个键值对上的每个独立节点的时间。向量时钟在追踪聚合状态或是阻止重复发送数据等方面是方便的，但对于n个节点、m个参数，向量时钟需要$O(nm)$的空间，对于上千个节点以及数十亿个参数，存储和带宽是不可能的满足的。
但是，参数服务器基于范围的通信模式，很多参数都有相同的时间戳：如果节点在一个范围内push参数，那么这些与这个节点相关的参数的时间戳几乎是一样的。因此，它们可以被压缩到单个范围的向量时钟中。更特别地，假设$vc_i(k)$是节点$i$在key $k$的时间，考虑key的范围为$R$，则范围向量时钟$vc_i(R)=t$表示对于任意key $k\in R$，$vc_i(k)=t$。
开始时，每个节点$i$只有一个范围向量时钟，从初始时间戳覆以及范围0开始覆盖了全部的参数key空间。每个范围设置可能分开（split）范围并且创造至多3个新的向量时钟（见Algorithm 2）。这样对于$k$个独立范围、m个节点数，至多$O(mk)$个向量时钟。

algorithm2.png

3.2 Messages

• 节点可能发送信息给独立的节点或者节点群。信息包括一个在key范围$R$中键值对的列表以及与之相关的范围向量时钟：
  $[vc(R),(k_1,v_1),...,(k_p,v_p)] k_j\in R\ and\ j \in {1,...,p}$
  这是在参数服务器中最基础的通信格式，不仅用于共享参数通信还用于任务任务。对于任务通信，键值对可能采取（任务ID，参数（arguments）或返回结果）的格式。
• 信息可能包含在范围$R$中所有可用keys的子集。缺失（missing）的keys被分配相同的时间戳但不改变它们的values。
• 信息可以根据key的范围切割：当worker发送一个信息给整个服务器群，或者当key的接收节点的分配改变时，我们分割（partition）键值对列表并且分割（split）范围向量时钟。（类似于algorithm2）。
• 机器学习通常需要高带宽，因此信息压缩是必要的。

1. 迭代次数之间，训练数据经常保持不变，worker可能多次发送相同的key列表，因此接收节点缓存key列表是很有必要的，这样以后发送者只需要发送列表的hash而不需要发送列表本身。
2. values中可能包含很多为0的entries，而我们只需要非零的键值对。使用fast Snappy compression library压缩信息，有效移除0。

• key缓存和value的压缩是可以共同使用的。

3.3 Consistent Hashing

figure7.png

• 参数服务器划分keys与传统的分布式哈希表（DHT）做法相似：keys和服务器节点的ID共同插入到hash ring中（见figure7）。每个服务器节点管理的key范围从它的插入点逆时针方向到下一个点，这个节点就被称为这个key范围的master。一个物理服务器经常在ring中经常表示成多个“虚拟”服务器来提高负载均衡和恢复。
• 我们通过使用直接変换的分布式哈希表设计简化管理。server manager掌控ring的管理。所有其他节点缓存部分key到本地，这可以使节点直接判定哪个服务器是负责哪个key范围，并且在有任何改变时收到通知。

3.4 Replication and Consistency

figure8.png

• 每个服务节点存储一个副本包含$k$个逆时针方向相邻的key范围。例如figure7，$k$=2，server1复制属于server2和server3的key范围。？
• 工作节点采用push和pull与key范围的master通信，任何master上的修正都被带上时间戳复制到slaves中，对数据的修改将同步地push到slaves中。
• figure8：worker1 push $x$到server1，server1调用用户定义函数$f$来改变共享数据。push任务只完成一次，数据更改$f(x)$就被复制到slave。
• 参数服务器允许在聚合后复制来减少网络通信。服务器节点经常聚合来自工作节点的数据，如求和本地梯度，服务器可能延迟复制直到聚合完成。figure8右图，两个worker分别push $x$和$y$到server，server先将push聚合成$x+y$,然后运用改变$f(x+y)$,最后实行复制。

3.5 Server Management

• 为了实现错误容忍和动态规模，我们支持添加和移除节点
• 添加节点：

1. server manager 分配给一个新节点一个key范围作为master。这可能导致另一个key范围被分割或从一个结束节点中移除。
2. 节点获取范围内的数据作为master来维持（maintain），并且获取$k$个额外范围作为slave来保持（keep）。
3. server manager广播节点的改变。信息的收取者可能基于新的key范围收缩它们的数据并且重发未完成的任务给新节点。

• 从$S$节点获取在$R$范围内的数据可分为两阶段：

1. $S$预复制在范围内的所有键值对和与之相关的向量时钟。这可能导致一个范围向量时钟被分割（类似于Algorithm 2）。如果新节点在这一步失败，则$S$仍然为改变。
2. $S$不再接受影响key范围$R$的信息，同时，$S$发送所有发生在$R$范围在预复制阶段的改变给新节点。

• 节点N在接收到节点改变信息后，首先确认它是否维持key范围也是$R$。如果是并且这个key范围不在是N维持，则删除在$R$范围内的所有相关的键值对和向量时钟。然后，N浏览所有还未接收到回复的外出（outgoing）信息。如果一个key范围与R相交，则信息会被分割和重发。

3.6 Worker Management

• 增加一个工作节点W比增加一个新服务节点简单：

1. task scheduler 分配W一个范围的数据
2. 这个节点从网络文件系统或存在的workers中载入范围内的训练数据。然后，W从servers pull共享参数。
3. task scheduler 广播这个改变，可能造成其他workers释放一些训练数据。

• 当一个worker离开时，task scheduler可能启动复位。

4 评估