腾讯大规模分布式机器学习系统无量是如何进行技术选型的？

AI 前线导读：在互联网场景中，亿级的用户每天产生着百亿规模的用户数据，形成了超大规模的训练样本。如何利用这些数据训练出更好的模型并用这些模型为用户服务，给机器学习平台带来了巨大的挑战。腾讯开发了一个基于参数服务器架构的机器学习计算框架——无量框架，已经能够完成百亿样本 / 百亿参数模型的小时级训练能力。无量框架提供多种机器学习算法，不但能进行任务式的离线训练，还能支持以流式样本为输入的 7＊24 小时的在线训练。
更多干货内容请关注微信公众号“AI 前线”，（ID：ai-front）

1. 背景

QQ 浏览器首页的推荐 Feeds 流。业务入口如图所示：

图 1 QB Feeds 流业务

浏览器的 Feeds 业务每天的流点击曝光日志在百亿级；为了更好的给用户提供个性化的推荐服务，如果我们取半个月的数据来训练推荐模型的话，则我们会面对一个千亿样本的状况。

图 2 模型在线表现的时间衰减曲线

此外，对一个训练好的模型，我们观察了模型在线的指标变化，如图所示。这个图说明我们的 Feeds 流业务是一个时效性高度敏感的业务，在线用户访问的规律实时在变化，要取得最好的业务效果，我们必须不断及时的更新模型。浏览器另一个业务——识花君，需要用百万级图片预训练一个多分类的图片分类模型，如果采用单机单卡的模式，大约需要半个月才能训练一个收敛的模型；如果使用 TensorFlow 的分布式训练也大概需要一周，有没有更高效的方法呢？

针对这两个业务场景，接下来我们做一些技术分析，看看有没有一些解法。

场景 1：大数据 + 大模型

在可以低成本获得样本的场景，比如广告、Feeds 流的 ctr 预估场景，因为不需要标注我们就可以低成本的获取海量的正负样本，这就会促使我们设法从这海量的样本里学习足够的知识。

什么样的算法模型可以从海量数据里学习充分的信息呢？这里从 VC 维理论出发，我们知道一个模型可以容纳的信息是有限的；下图概括了样本数量、模型规模和模型效果之间的关系，这里我们用模型效果来侧面反映模型容纳的信息量是基于这样的假设：如果一个模型从同样规模的数据里学习到了更多的信息，那么我认为它在业务上会体现出更好的效果。这个假设当然还会有很细微的条件，但这里就不深究了。

图 3 模型信息和样本、模型规模的关系

从该图我们可以直观得出一个结论，对于可以轻易获取海量样本的场景，我们需要用足够大的模型去容纳其中的信息。为避免过于直观，这里我且举一个例子，以一个亿级 Feeds 流业务为例，如果每天用户点击超过一亿，那么单天用户的 pv 可能在 5-10 亿甚至更多；如果我们取半个月的数据来训练一个 CTR 预估模型，涉及到的样本量在 200 亿左右（不考虑向下负采样先），而如果我们的模型参数是样本的 10 倍的话（这个范围并不夸张），我们的模型参数数量在 2000 亿，每个参数用四字节表示，我们的模型将达到 1TB 左右；而如果我们用 double 精度则接近 2TB。

这个量级的模型如何训练？如何做在线 serving？2000 亿的原始样本如何存放？答案是唯一的：我们需要一个分布式系统。

场景 2：大数据 + 中小模型

这是另一种场景，以某图像分类业务为例，我们要将一个标注好的图像数据集通过模型分类到几千个类目上。数据集我假设 1000w 张图片；咋一看，似乎这个和大数据关系不大，才一千万而已，但注意这里是图片，如果我们把图片的每个像素作为一个样本来对待，这个数据就大了；为什么这么说？因为我们用 CNN 类的网络来训练的时候，图片本来就是以像素输入；是的，这里的大数据其实想表达的是对算力的要求。

如果我们在单机单卡（GPU）上来训练这个分类模型（以 resnet-101 为例），可能需要 2-3 周；真的是“洛阳亲友如相问，就说我在跑 training”。对于算法同学来说，如果我们要等一个模型结果需要 3 周，这显然是很让人沮丧的一件事。那么我们有没有机会把这个时间缩短到天甚至小时级别呢？答案也是一样的：我们需要一个分布式系统。

上面两个场景也许只是鹅厂众多业务场景中的一小部分，但我相信是有一定的代表性的。这里共同的答案是我们需要一个分布式系统来应对业务场景带来的工程挑战。从机器成本的角度，我们不太可能去定制能满足需求的单台机器来解决；从人力成本的角度出发，我们也不太可能容忍模型训练速度的超级低效；因此使用相对便宜的机器构建一个面向机器学习需求的分布式系统是我们唯一的选择。

2. 分布式机器学习的架构与物理设计

分布式机器学习系统，顾名思义，和分布式文件系统、分布式后台服务类似，是一个分布式系统（这似乎是废话）；再结合机器学习就不一样了，这是一个面向机器学习场景的用相对便宜的机器组建的分布式系统（这还是废话）。那么和传统的分布式系统相比，分布式机器学习系统有哪些独有的特点呢？做这类系统的开发需要哪些算法知识和工程思维呢？

和传统的分布式系统很大的一个不同的地方在于，传统的分布式系统是 operation-oriented；以存储系统为例，传统分布式文件系统是绝对不能接受比如一个数据块写错地方了这样的事情的。

图 4 operation-oriented system

与之不同的是，如果我们以 operation-oriented 的要求来应对分布式机器学习的问题的话，那结果会是相当悲剧的，以我们目前的算力，我们可能根本没法在可接受的时间内完成一个大模型的训练的。然而上帝关上一扇门的时候也许会帮你掀开屋顶；机器学习的模型和算法本身都是有充足的容错能力的，你丢个样本，或者丢个梯度基本不影响模型的最终收敛，而这给了分布式机器学习系统一条出路，我姑且称为 convergence-oriented system。

图 5 convergence-oriented system

如图 3 所示，convergence-oriented 系统和下山比较类似，下山的路有无数条，中间你走偏了也无所谓，只要你的大方向是往山下即可。

因为机器学习算法自身的特点，分布式机器学习系统相比于传统分布式系统在数据通信、同步协议、容灾等方面都有极大的活动空间，也为我们追求极致的性能打下了基础。对分布式机器学习系统的通信、同步协议有兴趣的通信可以参考之前的拙作 [1] 和 Eric 的相关文章。接下来带大家游览一下应对两种场景的可用的系统架构！

2.1 参数服务器

关于参数服务器，之前的拙作 [1] 已经有较多的论述，这里不再详细展开，仅作简单介绍，想深入了解的同学请根据 [1] 按图索骥。

图 6 参数服务器架构示意图

如图所示，参数服务器逻辑上分为 server 和 worker 两类角色；server 负责存储模型参数，每个节点负责一个参数分片；worker 负责根据不同的数据分片来计算该数据分片涉及到的参数对应的梯度增量，并回传给 server 节点以 update 模型。因为数据和模型都是分布式存储，架构简单健壮，理论上该架构可以支持的模型规模是无限的；但是另一方面我们也应该看到，因为每个数据分片涉及到的参数分片可能分布在不同的机器上，导致我们每增加一台机器，网络的整体传输量会有所增加；如下图所示

图 7 参数服务通信示意图

因此，在参数服务器架构下，相对于算力瓶颈，网络更容易成为我们的瓶颈，而这又该如何解决呢？请继续往下浏览。

2.2 ring-allreduce

对于图像分类、机器翻译这类强依赖 GPU 机器的场景，我们来看看另一种情况：

图 8 使用 GPU 构建的 ps 架构

如图所示, 如果我们使用 GPU 搭建一个 PS 集群，我们将面临更为严峻的挑战；因为 GPU 的运算速度极快，我们在做参数 reduce 的时候，与 GPU0 的通信时间将成为整个系统的 dominant time 而让系统中的 GPU 心有余而力不足。为此，百度的 SVAIL 团队 [3] 从高性能计算领域借鉴了 ring-allreduce 思想，构建了分布式机器学习的 ring-allreduce 架构，如下所示：

图 9 ring-allreduce 架构示意图

将 GPU 布置成环状现在以有官方组建 NCCL 可以支持，对 NCCL 原理感兴趣的可以参考 [4] 等相关 paper。

图 10 使用 NCCL 搭建的多机多卡环

如图所示, 当我们使用 NCCL 将多台 GPU 机器搭建成环状结构时，我们可以看到在换上以此传输的话网络带宽可以得到比较充分的应用。接下来解释下 allreduce 的概念，一般的 reduce 概念如下

图 11 reduce 操作语义

而 allreduce 的概念如下

图 12 allreduce 操作语义

因为上述图片已经足够直观，这里就不多加解释了。接下来我们介绍 ring-allreduce 为什么适合 GPU 集群数据并行的场景；考虑到中小规模的模型我们可以存放在单台机器上（单卡 or 多卡但不跨机器），每台机器根据自己的数据分片训练模型后通过环状通信来做 allreduce 操作；这样的设定下整个系统的网络通信量不会随着机器增加而增加，而仅仅与模型和带宽有关，相对于参数服务器架构而言，这是极大的提升。详细的推导过程可以参考 [5]，我就不赘述了。实际的网络通信流程如下所示

图 13 Ring-allreduce 通信的物理过程

在介绍了参数服务器和 ring-allreduce 两种不同的分布式机器学习的系统架构以后我们该如何根据自己的业务场景来合理的选择架构、算力社保、部署策略呢？请看下节

2.3 物理实现的设计选择

前两节介绍的两种逻辑架构在物理实现的时候可以有多种选择，这里做几种推演：

2.3.1 PS 数据并行

仅使用 PS 架构来支持数据并行，如下图所示

图 14 单 server 参数服务器架构

这种架构下仅仅支持 worker 对数据进行并行计算，模型存放在集中的 server 节点，和 spark 的架构类似。因为是单节点，所以模型不可能太大，因此这个模型仅仅对照意义多一点，实际上基本不会这么用。

2.3.2 PS+p2p

在实现的时候，将参数服务器的 worker 和 server 两个角色融为一炉，在一个进程里既有承担 server 角色的线程，又有负责 worker 的线程；因为 worker 以计算为主，server 以参数存储为主，这种融合有一定的合理性，如下图所示，虚线框表示一个物理进程，一台机器上可以部署一个 or 多个这样的物理进程。

图 15 P2P 结构的参数服务器

这种架构的不足之处在我看来有两点：1. 角色耦合，较难根据机器来调配线程比；debug 也相对困难一点；2. 架构耦合，扩展的灵活性较差；调度系统交护模块、监控模块的配合、灾难恢复都有一定的风险。

2.3.3 PS 角色分离

与图 13 不同，如果我们将 worker 和 server 两个角色实现为解耦开的两个独立进程，在可以给调度系统流出更多的活动空间。同时对架构的扩展也预留了空间，如果我们再独立一个单独的调度模块出来，则演变为下一种架构。

2.3.4 PS+scheduler

当我们将 worker 和 server 拆成两个独立的模块，并引入一个 scheduler 模块，则会形成一个比较经典的三角色分布式系统架构；worker 和 server 的角色和职责不变，而 scheduler 模块则有比较多的选择：1. 只承担和下层资源调度系统般若（类似 yarn、mesos）的交互；2. 除 1 外，额外增加对 worker、server 心跳监控、流程控制的功能；如下图所示：

图 16 带控制模块的参数服务器

引入 scheduler 模块的另一个好处是给实现模型并行流出了空间，关于模型并行概念的理解，请参考 [1]；关于在 scheduler 模块下如何实现对模型参数的调度以达到模型并行的效果，请参考 [6] 中对 SchMP 编程范式的论述；调度模块不仅有利于实现模型并行训练范式，还有其他好处；比如通过针对特定模型参数相关性的理解，对参数训练过程进行细粒度的调度，可以进一步加快模型收敛速度，甚至有机会提升模型指标。这块也是一个很值得探索的方向, 有兴趣的同学可以进一步参考 [7]。熟悉分布式系统的同学可能会担心 scheduler 模块的单点问题，这个通过 raft、zab 等 paxos 协议可以得到比较好的解决，无需过于担心。

2.3.5 ring-allreduce+PS

初始的 ring-allreduce 有一个开源版本是 uber 实现的 horovod 框架，通过测试我们重现了 horovod 论文里的加速情况，如下图所示

图 17 多机多卡场景下 ring-allreduce 架构加速比对照 TensorFlow 加速比

从该图可以看出 ring-allreduce 的加速比和理想加速比的斜率几乎完全一致，而 TensorFlow 的加速比则远低于次；这证明了 ring-allreduce 通信机制相对于 ps 机制在网络通信方面的优势；但与 PS 架构不同的是，初始版本的 ring-allreduce 假设模型参数需要单卡可以存下，另外如果模型中全连接层比较多，则全连接层的强耦合性结合 allreduce 类似 bsp 的同步机制，还是会让网络通信时间成为瓶颈。因此，在 ring-allreduce 环境下，我们是否可以做模型分片、同步协议的改造，比如利用 SSP 来替换 BSP，或者利用梯度压缩来加快 allreduce 进程都是值得探索的方向。

3 技术成果

经过大半年的封闭开发，目前无量系统已经支持了 LR、FM、FFM、DNN 的离线训练和在线实时训练。支持了 FTRL、SGD、Adam、AmsGrad 等多种优化算法。针对不同的优化算法，我们在梯度压缩上也做了一些基本的尝试，如图所示

图 18 不同优化算法做梯度压缩后的收敛指标对比

如图所示，在 LR 算法分布式训练过程中，我们过滤掉 99% 的梯度，仅传输剩下的 1% 的梯度依然可以达到模型收敛的效果；而且指标可能还略有提升，我们推测可能是大范围过滤梯度引入了一些 regularition 的作用。

除了常规算法之外，我们自研了大规模 embeding+DNN 的分布式训练支持，如图所示

图 19 自研分布式 DNN 模型

该模型在召回和精排环节都可以应用，目前已经在召回环节灰度。回到最开始的问题，我们封闭开发无量的一个初衷还是为了支持 Feeds 业务精排环节，那么面对大数据 + 大模型我们现在是什么情况呢？无量支持了千亿级特征空间的稀疏 LR 的分布式训练；目前在线已经实际使用到百亿特征，百亿样板，训练好的模型为了方便在单机上做 inference，我们会做一些裁剪；详细过程我可以参考我另一篇分享。

图 20 在线效果提升比例

使用了基于无量系统训练的模型之后，Feeds 在线 CTR 和曝光效率都有显著的提升，如图所示；相对提升百分比在两位数，这个提升是在基于 GBDT+ 细粒度特征的粗排基础之上的提升，因此这个结果还是非常符合业务的预期的。

在另一个方向上，我们基于 ring-allreduce 的架构，对大数据 + 小模型的 cv 场景已经可以做到小时级模型输出；该场景以后会做更深入的探索。

4. 团队介绍及文章计划

无量项目是 MIG 移动浏览产品部与无线运营部联合开发的，团队主要开发成员由大数据中心下的智能应用组、运营部下的计算框架组以及浏览器大资讯业务相关同学构成，主要成员如下：robertyuan、suziliu、clarebu、yancyliu、wahmingchen、burnessduan、binzhu、、williamqin、carbonzhang、janwang、collinhe、joeyzhong、foxchen、brucebian 等。

本篇为系列分享的第零篇，主要介绍分布式机器学习框架的背景及可用架构；接下来我们会从系统整体概况、工程挑战、算法挑战、业务应用等角度展开系列分享，敬请期待！

致谢

特别感谢浏览器和运营部两位老板 henrysxu 和 xinliu 的支持，没有老板的支持我们不会有机会去探索分布式训练这个领域；感谢 foxchen、taydai、brucebian 的给力支持，使得项目的进展过程中，资源的支持始终走在开发先列。感谢 rainyu、joeyzhong、janwang 的支持，在过程中对项目高度关注，经常组织大家讨论和勾兑；最后感谢 robertyuan、suziliu、clarebu、yancyliu、wahmingchen、burnessduan、binzhu、hbsun 等同学的辛苦开发，过程中有过碰撞，最终时间让我们了解彼此，共担重担！

还有很多同学在项目上线过程中提供了极大的帮助，如 larrytu、aiyima、ballwu 等和我们一起对流程、对参数，可能无法一一列出，然感激之情，不减毫厘！

引用

[1] 大规模机器学习框架的四重境界
[2] More Effective Distributed ML via a Stale Synchronous Parallel Parameter Server
[3] https://www.sohu.com/a/127596575_494939
[4] Bandwidth Optimal All-reduce Algorithms for Clusters of Workstations
[5] https://www.zhihu.com/question/63219175/answer/206697974
[6] STRADS: A Distributed Framework for Scheduled Model Parallel Machine Learning
[7] Managed Communication and Consistency for Fast Data-Parallel Iterative Analytics