一、背景
推荐和广告领域已经大规模应用DNN模型，但大规模稀疏性仍然是该领域模型的本质特点，为了增强模型的拟合能力，模型稀疏参数会达到万亿规模，单模型的物理大小超过1TB，需要分布式的服务来承载；另外模型在线预估场景每秒的参数获取量由每秒用户请求量（大于1000）,每个用户请求计算的item数量（大于1000），每个item获取的参数量（大于200）的三者乘积组成，其对应每秒的参数获取量KPS（Keys Per Second）在2亿的级别，即使用20个副本来分摊流量单机KPS也超过1000万，行业内还没有一个通用key-value存储引擎（Redis/RocksDB/Pika[1]）能满足上述的两个核心诉求，于是BIGO结合模型在线预估场景的特点进行了存储引擎和分布式服务的定制设计，获得了万亿参数规模下高性能参数服务的能力，取名Cyclone。
二、整体架构设计
1、核心目标
1、单机能够支撑1000万以上KPS且latency低于10ms
2、通过分布式集群支持万亿参数规模，TB级别的模型
3、支持多副本和多个shards的一致性增量更新
4、系统可用性达到99.99%
2、架构图
上图为整体系统架构图，在线服务部分包含cyclone proxy、cyclone meta、cyclone server三个核心服务，参数数据存储在Cyclone Server中。
3、设计思路
（1）底层存储引擎的核心目标是需要极致的查询性能，同时考虑到模型在线预估场景不会实时做insert/update/erase操作，只需要整体更新模型版本，因此我们采用完美hash（cmph[2]），在离线阶段build好索引结构，在线服务阶段直接mmap数据即可提供查询，类似于FlatBuffers[3]的“Access to serialized data without parsing/unpacking”；此外cmph可以最大限度的压缩索引部分（平均每个key大约用3个bits来表示），同时又可以保证绝对O(1)的查询效率，value部分整体单独存放，通过offset来获取value的值
（2）整个集群由一组cyclone meta服务来做模型管理，负责模型的发布、迁移、下线；meta服务采用单master多slave架构，且master一旦失效可以快速从slave中选择出新的master承接服务；并由etcd集群存储模型元信息，保证元数据的可靠性
（3）cyclone server负责模型参数存储和查询，且通过heartbeat和master cyclone meta交互，一旦server失效，cyclone meta可以快速进行服务迁移保持各模型具有稳定的副本服务能力
（4）对外由cyclone proxy层提供统一访问接口，其内部会负责模型的路由、keys的dispatch与merge、各cyclone server的load balance，同时该层使得模型副本变化以及分片迁移等内部情况对外透明
三、关键技术点接拆
1、模型拆分
首先，把模型整体拆分成若干shards，这部分需要充分考虑feature keys的特点，我们希望sharding后每个分片里的keys数量尽可能近似相等。一般来说shard_idx = hash(key) & mask， mask=2^N–1。至于hash函数，考虑到这里的feature key已经是uint64_t，如果本身不够均匀，可以尝试用一些mixer[4]。
其次，每个shard内部，为了避免单个cmph数据量太大（1000万key的build时长大约为15s）导致build时间太久，可以进一步拆分成多个sections，每个section都是一个独立的cmph模型，建议控制在1000万左右的keys，这里的拆分也需要用到hash函数，建议与sharding过程用的一致。
2、模型更新一致性
在模型更新期间，我们希望模型的所有shards、所有副本能够同时完成版本切换，避免步调不一致带来的预估效果损失，因此采用两个阶段来完成版本切换：第一阶段是cyclone meta控制多个cyclone server完成模型shards的download与load/init；第二阶段是cyclone meta通知各个cyclone server开始serving，同时在元信息里完成模型的版本切换，在模型升级期间会有两个版本的数据同时存在于cluster内。
3、水平扩展
cyclone proxy是无状态的对等节点，因此可以任意水平扩展；cyclone server被cyclone meta调度，新加入的cyclone server节点会被逐步的把模型调度上来，同时由于模型是多副本，因此即便某台cyclone server失效也并不会导致某模型数据不可访问，cyclone meta会把失效的cyclone server上的模型分摊到其他cyclone server来补齐各模型应有的副本数；cyclone meta将模型的元数据（例如已经发布成功的模型、正在发布的模型、正在迁移/卸载的模型信息）持久化到etcd中存储，同时基于etcd的分布式锁，所有cyclone meta竞争锁，获得锁的meta自动成为master，提升为master的节点会从etcd获取所有模型的元数据，继续负责整个cyclone cluster的模型管理。
4、自适应调整模型副本数
cyclone meta会统计单个模型最近一段历史版本的访问情况，根据查询量来动态调整新版模型的副本数；同时，如果当前版本在serving期间查询量增加太快，cyclone meta也会实时的增加副本来应对；此外，如果某个模型长期没有被查询，超过一定时长后会被判定为zombie模型，进而将其副本数逐步调小至0（即自动下线该模型）。
5、增量发布
为了更好的解决增量问题，我们需要索引内核可以支持在线批量insert/erase，为此我们在cyclone二期替换掉了cmph索引内核，改用自研的zmap内核（基于parallel flat hash map[5]改进的一种hashtable），zmap支持多读单写，可以在线更新增量数据，虽然索引部分相比于cmph会增大一些，但该索引格式支持在线更新，且查询效率要优于cmph，离线build比cmph更快。
Zmap的高性能主要得益于基于AVX2指令集能够单次扫描hash table索引部分32个bytes，进而可以让hash table的load factor提升到0.9以上，并通过精心设计数据结构使得在多读单写情况下基本达到无锁。有关zmap我们后续会在另外的文章里进一步详细讨论它。
现阶段cyclone二期已经上线，单模型可以在10分钟的粒度增量更新，大大提升了时效性。
6、cyclone性能
由于cyclone集群可以任意水平扩展，模型也可以通过更多的副本和shards来提升单模型的查询效率，因此我们重点来看下cyclone单机服务能力。选用一台线上服务器，1T物理内存，CPU为44 core，2.1G Hz，加载1000亿模型参数，单次mget查询的keys数量从1k递增到20k，分别观察其QPS和Latency。
可以看到，mget 1k单机可以达到2.1万的QPS，即单机2100万的KPS，随着单次mget的keys数量增加，QPS相应下降，但总的KPS基本是稳定在2000万以上；Latency方面，即便是单次mget 2万keys，客户端也能够在10ms内拿到结果。Cyclone server侧的cpu使用率为1100%左右，还远没有达到cpu资源瓶颈，之所以KPS无法继续提高是因为网卡吞吐量已经达到极限。
从性能测试看Cyclone完全可以达到万亿参数规模的模型预估的性能要求。
四、总结与展望
cyclone除了为DNN提供配套的模型参数查询，其实也可以给任何service提供高吞吐的key-value查询，因此cyclone已经被广泛用在了user-cf、item-cf、ad service等各种场景的服务中。
在深度学习日益普及的今天，相信很多公司都有类似的场景需要类似cyclone这样一套支持水平扩展的、高可用的分布式服务来提供高性能模型参数查询，后续我们会进一步持续打磨cyclone，等到更加成熟的时候BIGO将对外开源。
References：
[1] https://github.com/pika/pika
[2] http://cmph.sourceforge.net/
[3] https://google.github.io/flatbuffers/
[4] http://zimbry.blogspot.com/2011/09/better-bit-mixing-improving-on.html
[5] https://github.com/greg7mdp/parallel-hashmap
版权声明
转载本网站原创文章需要注明来源出处。因互联网客观情况，原创文章中可能会存在不当使用的情况，如文章部分图片或者部分引用内容未能及时与相关权利人取得联系，非恶意侵犯相关权利人的权益，敬请相关权利人谅解并联系我们及时处理。
关于本文
本文首发于公众号【BIGO技术】，感兴趣的同学可以移步至公众号，获取最新文章~
转载：https://blog.csdn.net/Sprite_CJ/article/details/112391406