万字干货 | 荔枝魔方基于云原生的架构设计与实践

近年来，荔枝集团在国内和海外的业务迅速发展，业务数据规模也是成几何式地增长，海量数据的计算分析场景、业务智能算法应用需求随之而生，为了快速地满足业务发展的需要，我们面临着诸多的技术挑战。

技术挑战

工程问题

资源问题

1、计算资源存在滥用

如生产环境、预发环境上的在线服务独占GPU，但是在很多业务细分场景下GPU资源使用率低，特别是在一些预发环境下经常出现偶尔才有请求的情况。

2、资源在时间维度利用率较低

有很多团队下的训练是单机多卡的模式训练，任务之间的训练无法跨越单机的限制，任务之间的训练靠人工去控制资源，多台机器在同一时刻无法达到最大化利用率，比如A机器上的任务把机器资源已经跑满负荷了，B机器上当前可能资源剩余很多。

3、GPU机器当作CPU机器使用

超大型CPU任务在GPU机器上跑或者是GPU任务流中的大型CPU计算过程在GPU机器上执行（占用磁盘、机器网络资源、GPU任务计算过程中所需要的CPU及内存资源等等）。

4、资源环境运维困难，机器迁移扩容效率低下

很多时候开发人员的任务开发及运行环境与固定机器绑定，如果机器出现损坏、扩容机器等这些复杂的环境都需要重做一遍，有时出现几天甚至更久才能交付一台机器，同时单机资源有限在大型算法模型情况下经常会出现如磁盘、内存等资源不足。

5、业务开发人员对代码及架构的优化经验、意愿不高，导致资源无法有效利用，机器成本上升快于业务发展

大部分情况下相关算法人员对资源利用优化经验或者意愿并不高，基本上是通过加机器来满足计算资源的不足，但有时候通过技术架构和代码的小小优化能节省大量的资源成本。

技术问题

1、数据量大，模型大单机无法支撑，同时模型训练周期过长，线上模型线新缓存，影响业务。

2、单模型训练周期长，无法进行超参搜索，模型参数优化难度大，工作效率低下。

3、算法框架多，不同团队对算法框架要求不一样，如TensorFlow、Pytorch、Sklearn、XGBoost等等，各自版本要求不一样，环境复杂，运维难度和工作量极大。

所以我们需要解决如下一些问题——

资源统筹与边缘计算

1、统一资源管理，将通过人工管理资源改变成机器进行资源管理，平台管理资源申请及资源调度，对不合理资源申请进行智能修正，同时采用合理的智能的调度算法对行计算资源调度，使资源达到最大化利用。

2、边缘计算，在公司业务全球化的大背景下，在国内不同的地区、国外不同的国家都有业务，如果把所有数据传输到一个中心机房进行计算的话数据传输压力会比较大，计算能力比较低下且数据安全性存在一定不足，而如果在每个地方维护一个中心计算集群的话建设成本高，所以我们需要边缘计算，在中心机房执行计算命令，数据与计算在边缘机房进行，以提升计算效率，降低计算成本。

3、对GPU进行虚化，减少GPU资源浪费。

提供一站式模型训练能力

1、开发、运行等环境模版化，达到只需几秒钟就能复制一套新环境。

2、提供便捷的开发工具，开发人员可以快速进行环境安装制作及代码开发、调试、发布上线。

3、提供任务流编排功能，使用者通过组件拖拉的形式即能编制出一套复杂的训练流程
提供训练任务定时调度功能，支持各种的定时任务操作（补录、忽略、重试、依赖、并发限制）。

将复杂技术模版化、组件化

1、提供超参数搜索组件，算法同学能方便快捷进行超参数调优，提高工作效率。

2、提供模版化如

Pytorch/TensorFlow/XGBoost/Spark/Ray/Horovod/Volcano等分布式训练组件，使用者通过简单的拖拉组件就可以完成复杂的分布式计算过程。

魔方智能计算平台介绍

荔枝魔方智能计算平台面向于人工智能、大数据开发人员使用。集大数据计算、算法模型训练、任务调度、代码开发、资源调度、边缘计算等功能于一体，为推荐、搜索、风控、广告、数据分析、数据应用、智能对话等提供能力支撑。

架构设计

技术选型

在机器学习领域，大家可能接触到最多的有Airflow/MLflow/Kubeflow等等，除了MLflow和Kubeflow之外的大部分开源框架都只是偏向于任务流编排及任务定时调度的，对于机器学习相关的支持没有或者是很弱，其中的MLFlow在机器学习领域内应用比较还是比较多的，但是MLFlow只适合于小规模团队与小规模的模型训练，对于大型分布式计算、资源统筹调度等等支持还是比较弱。

Kubeflow是由Google开源的框架，Kubeflow 旨在通过提供一种直接的方式将用于机器学习的同类最佳开源系统部署到各种基础设施，从而使机器学习工作流在 Kubernetes 上的部署变得简单、便携和可扩展，同时有两大 IT 趋势开始升温——云原生架构的主流化，以及对数据科学和机器学习的广泛投资，Kubeflow 完美地定位于这两种趋势的汇合点。它是云原生设计，专为机器学习用例而设计。

基于前面已经介绍过了我们的痛点及需要解决的问题点，通过Kubernetes 的原生架构更好更快的集成开源组件运用到机器学习平台中，以满足业务的需要，如与Volcano集成能更好的进行资源调度，基于Kubflow提供的Train-operator快速搭建起TensorFlow/Pytorch/MXNet等分布式训练能力，基于Kubernetes我们能在上层打造更贴合用户的功能，如训练机器创建与销毁，用户只填入简单的资源需求后台就能秒级的创建出一套用户所需要的新环境出来供使用者进行开发、测试、模型训练、模型发布上线等等。

但是Kubeflow也存在着很多的不足

1、任务流构建使用复杂、需要通过Python脚本构建任务、同时设置TASK运行过程中的参数复杂，比如磁盘挂载、资源配置等等，使用成本比较高。

2、缺少环境开发相关组件，比如用户自定义的开发环境的开发工具等等。

3、无法集成除Kubeflow支持以外的组件，比如集成Spark/Ray/Volcano，还比如对GPU虚化后无法应用到平台当中去。

4、无法按分组进行资源调度管理，比如按线上机器集群进行资源调度，按训练集群进行资源调度、按项目分组进行资源调度等等。

我们的选择

以Kubeflow为平台的基础，在Kubeflow的上层我们进行封装及扩展，打造集团统一计算平台服务于集团国内和海外算法模型计算相关业务。

技术架构

资源管理

硬件层主要是实体机器上的资源，比如磁盘、GPU\CPU等等资源管理主要是利用Kubernetes进管理集群的资源，在存储方面选择使用Ceph来管理集群中的存储资源。

Rancher

为了降低Kubernetes的安装及维护的复杂度，我们基于Rancher来搭建及管理K8S集群。

Rancher不仅可以集中管理部署在任何基础设施上的Kubernetes集群，还可以实行统一的集中式身份验证和访问控制。由于无法确定资源运行的位置，我们可以轻松地在不同的基础设施之间调用集群，并在它们之间进行资源迁移，同时更方便于K8S集群的扩容、升级、运维等等。Rancher 中文官网地址：https://docs.rancher.cn/docs/rancher2.5/overview/_index

Ceph

虽然Ceph的读写性能并不高，大概只有50M-60M/秒，相对于CFS等等性能有一定距离，在立项前期发现Ceph在K8S中安装比较方便简单，能很快的集成到系统中来，还有就是Ceph通过系统挂载后，用户能像访问本地文件系统一样访问Ceph集群上的文件，使用起来也方便简单，因而当时考虑利用Ceph来放置训练任务中的配置文件及需要执行的代码，这样分布式下进行训练会变得更简单方便，训练数据可以放置在HDFS等等之类的存储集群上，这样50M-60M/秒写性能完全能满足需求。

Kubeflow

Kubeflow中我们主要使用到了KFP/Argo/Katib/TrainingOperator/TensorFlowboard等组件。

Argo

Argo是一个开源原生容器工作流引擎，用于在Kubernetes上开发和运行应用程序。Argo Workflow流程引擎，可以编排容器流程来执行业务逻辑。

KFP

Argo 已经有了一整套任务流处理流程了，KubeFlow为什么还要在Argo上进行再次开发呢？首先就是Argo中的数据都是保存在ETCD中的，但是ECTD中的数据有大小的限制，数据总大小及每条记录大小在ETCD中都是有限制的，但是像流程模板，历史执行记录，这些大量的信息很明显需要一个持久化层（数据库）来记录，明显ECTD是不能满足需求的，这样就需要对这些功能进行增强，同是在ML的领域的用户界面层，KFP也做了较多的用户体验改进。包括可以查看每一步的训练输出结果，直接通过UI进行可视化的图形展示。

分布式训练在机器学习中是一个不可缺少的部分，模型训练大都伴随着大量的数据需要进行计算，单机的资源往往是有限的，利用多机资源分布式进模型训练是加速模型训练的一个重要的手段。

核心功能解析

自定义开发环境-秒级创建Jupyter开发环境提升工作效率

(Jupyter界面图)

为什么需要Jupyter+自定义容器环境：

1、提供易用的IDE工具，辅助开发，提高开发效率。（在很多的开发测试环境依赖于Linux等机器环境，很多情况下的同学要么通过Linux无图形化界面进行开发，或者本地开发，然后再上传代码、配置等，这样的开发效率非常低下）

2、隔离用户空间，减少开发过程中相互影响。

3、隔离环境，满足不同开发需求对不同环境的要求。（如一些库对GCC版本要求较低，有些库要求高，他们之间相互影响）

4、对环境做镜像，达到秒级创建新环境，相比起在实体机器上重建环境少则一天多则可能一周都搞不定一个复杂的环境。（比如机器迁移、故障等等环境重建，比如工作交接、新同学入职开发环境搭建，只要一键化就可以做到）

我们通过Docker将用户的开发环境进行隔离，针对于不同种类型的环境构建出一个基于Jupyter基础镜象的开发镜像环境，这样使用者就可以通过平台选择一个自已相适应用开发镜像，一键创建出一个容器环境，用户只需要通过Web页面就可以打开Jupyter进行代码开发了。这样做到环境的隔离，也能做到用户之间开发空间的隔离，每个用户都可以创建自己的Jupyter容器，大家开发上互不干扰，各种环境的依赖之间也是互不干扰，如果机器迁移或者机房迁移，用户只需要一键重建环境就可以了。

对于前台用户只需选择镜像、填写机器要求，比如CPU\GPU\内存\磁盘信息后就能创建一个环境，对于后台来说要解决的问题如下：

1、根据Jupyter的镜像，创建一个pod。

2、构建Jupyter启动脚本，在容器启动时将Jupyter的进程启动起来。

3、Pod启动后用户需要能够访问到这个Pod中的Jupyter，所以需要构建一套网络访问的服务或者叫CRD，最后将让Jupyter的访问地址能够在办公网络进行访问。

首先我们来看一下整个机器学习平台的网络访问结构——

从外部网络需要访问到K8S内部的服务需要通过外部的负载均衡负载到K8S的一些结点上，这些结点绑定着一个静态的端口（Nodeport），通过这个端口能将请求通过kube-proxy转发到对应的istio-ingressgateway最后由Istio配置的网关及VirtualService将流量转到对应的service上，最终通过service后就通访问到容器中Jupyter的服务了。

Gateway的配置是静态的，平台为了保证每个用户创建的每个Jupyter Pod都能独立进行访问，所以需要针对每个Jupyter Pod 动态创建Service\VirtualService进行绑定，最终达到可以动态创建Jupyter的效果。

分布式存储-分布式训练的基石

在分布式训练过程中，训练的容器资源是由K8S进行调度分配置，工作容器被分布在集群中的哪一台机器使用者是预先不知道的，这样我们就需要有一种介质来存储训练过程中所需要的代码、配置、数据等等，以便于在训练过程中任何一个容器都可以访问它。

在系统框架中已经介绍过了，平台采用的是Ceph为平台的分布式存储，同时与rook进行集成部署在K8S上，Ceph包含了包括对象存储、块设备、文件系统，显然这三种模式中文件系统存储便适合平台的使用方式，主要有如下几个原因：

1、Ceph文件系统能通过系统内核的方式进行挂载，使用者能像使用本地文件系统一样访问分布式文件系统，对于使用者来说无感知，使用成本几乎为0，对于那种以前都是单机模式开发的程序迁移成本会大大降低。

2、文件通过操作系统内核挂载，后期如果更换文件系统对于整个平台及平台的用户是无感知的，系统扩展方便。

平台按分类在分布式文件目录下创建子目录，同时按分类创建静态存储卷，比如用户空间存储目录：/xxx/xx2/user，会在K8S上创建一个PV及PVC，在woker容器创建时将容器下的目录mount到这个pvc上。

mount的目录主要分成几种模式，一种是用户级别的目录，这个目录下的文件只有用户。

自己可以访问，还有一种目录是项目组共享目录，这个目录是同属一个项目组下的用户才可以访问，另一种目录是全局共享目录，这个目录下的数据是所有用户都可以访问，每个运行的任务都会归属到个人、项目组，这样每个运行任务的容器在创建时都会将当前任务所归属的项目组、用户所属的目录挂载到运行容器中去。

解决了存储的问题后，我们就能在任何容器中像访问本地文件一样访问分布式文件系统上相同文件了，这样我们写一份代码，我们不用关心容器在创建在哪台实体机器上都可以进行访问了。

分布式训练-为百G以上级别数据进行模型训练护航

分布式训练基础知识介绍

本文所说的训练，指的是利用训练数据通过计算梯度下降的方式迭代地去优化神经网络参数，并最终输出网络模型的过程。在单次模型训练迭代中，会有如下操作：

首先利用数据对模型进行前向的计算。所谓的前向计算，就是将模型上一层的输出作为下一层的输入，并计算下一层的输出，从输入层一直算到输出层为止。其次会根据目标函数，我们将反向计算模型中每个参数的导数，并且结合学习率来更新模型的参数。

而并行梯度下降的基本思想便是：多个处理器分别利用自己的数据来计算梯度，最后通过聚合或其他方式来实现并行计算梯度下降以加速模型训练过程。比如两个处理器分别处理一半数据计算梯度 g_1、g_2，然后把两个梯度结果进行聚合更新，这样就实现了并行梯度下降。

训练并行机制

模型训练并行机制有三种，但是我们最常见的方式有2种：数据并行与模型并行，其中目前工业界中基本的训练框架实现都是基于数据并行的方式。

分布式训练最大的优势就是可以利用集群多机的资源，并行的进行计算，每一台机器承载着整个计算的一部分，也就是说一份大体量的工作由一堆人来做，每个人同时做其中的一小块事情，目前最常见的并行计算方式有2种：

模型并行：集运行的集群中，每台机器上计算着相同的数据，但是每台机器上运行模型中的不同计算部分。

数据并行：所有机器上的模型是相同的，但是需要训练的数据按机器进行拆分，每台机器计算数据中的一部分，计算完后再将结果进行合并。

目前工业界最主流运用最广泛的模式是数据并行计算。

数据并行的模型分布式计算实现架构

Parameter Server 模式

PS架构下所有的参数信息都存放在参数服务器中，参数服务（PS）在集群中可以是多台，Worker机器为工作结点，Worker结点首先从PS上获取参数信息，然后根据训练数据计算梯度值，计算完成后将计算的梯度更新到PS上，PS获取Worker过来的梯度值后对梯度求平均，最后返回给到Worker。

Allreduce 模式

AllReduce 模式是所有的机器上都具有相同的模型参数信息，每台机器计算一部分数据得到一个梯度值，然后执行 AllReduce 操作使得所有 node 结点都得到其它结点上的所有梯度值，最终更新本地的梯度值，AllReduce 每轮迭代都需要同步所有参数，对于网络来说是一个大的冲击，后来在2017年百度在Tensorflow 上实现了基于Ring Allreduce的深度学习分布式训练Ring Allreduce，大大减少了网络的压力。

参数服务器适合的是高纬稀疏模型训练，它利用的是维度稀疏的特点，每次pull or push只更新有效的值。但是深度学习模型是典型的Dense场景，Embedding做的就是把稀疏变成稠密。所以这种pull or push的不太适合。而网络通信上更优化的Allreduce适合中等规模的深度学习。又比如由于推荐搜索领域模型的Embedding层规模庞大以及训练数据样本长度不固定等原因，导致容易出现显存不足和卡间同步时间耗费等问题，所以Allreduce架构很少被用于搜索推荐领域。

分布式模型训练

上面介绍完分布式训练的一些基础知识后，我们来看平台是如何与这些框架结合进行模型训练，在机器学习平台上主要选取如下2种模式来支持深度学习模型的分布式训练:

基于RingAllReduce分布式训练

Horovod主要是基于Ring Allreduce的架构来进行分布式训练，Horovod 支持TensorFlow/Pytorch/MXNet等训练框架进行模型训练，在图像、音视频、文本等等分布式训练场景下使用非常广泛，对原框架（TensorFlow/Pytorch等等）的入侵很小，使用起来简单方便，对原代码做很小的改动就能进行分布式训练。

选择了使Horovod进行训练后需要有一套机制来组成Ring Allreduce通讯结构，可以看下图，这时我们需要有一套机制去创建容器，同时让他们组成一个环境环形的通讯结构。

我们首先来看一下Horovod的运行示例，如果是在实体机上执行的话只需要设置分布式下多台机器的SSH免登录，然后在其中一台机器上执行下面的代码，整个分布式就能正常的运行起来了。

但是在K8S上我们的容器是动态创建的，IP地址是动态变化的，执行完成或者异常后还需要对这一批容器进行回收等等操作，这时我们就需要一套这样的机制来实现上面说的这些功能，这时KubeFlow的MPI-Operator就能派上用场了。

MPI-Operator

MPI-Operator根据用户定义的CRD文件生成一个ConfigMap：

我们可以看到这个ConfigMap里边主要是生成了三部分，我们现在主要关注的是hosTensorFlowile和kubexec.sh，MPI-Operator会创建2种角色的容器: launcher、worker，这launcher在所有的worker容器启动后调用horovodrun命令，在上面官方文档中默认是通过SSH方式向集群中的其它容器发出执行远程命令，在launcher中MPI-Operator会设置launcher的环境变量OMPI_MCA_plm_rsh_agent。

这样最终在执行过程中会在launcher执行kubeexec.sh向worker发起命令执行用户脚本，同时MPI-Operator还管理运行过程当中成功与异常时容器的退出等等，这样在机器学习平台侧则需要构建MPI-Operator的CRD：

1: 构建文件挂载信息，将分布式存储挂载到Horovod的容器中去，以保证在任何容器中能访问到训练脚本代码和配置、训练数据等等。

2: 构建资源调度规则，如结点分配规则信息。（如如果有申请到GPU的资源，那则设置worker容器都分布到GPU的结点上去，如果只需要CPU资源则设置worker分配到CPU的结点上去，同时会按照平台的资源隔离策略，如资源有按照分组进行隔离将worker分布到当前分组所在的资源结点上去运行）。

3: 设置Pod之间的亲和策略，比如是GPU机器的话尽量将容器分布到相同的结点上，减少中间的一些网络损耗。

平台要解决的问题是通过上面一个简单的配置，就能实现复杂的分布式训练过程。

开始提交训练任务运行分布式任务：

CPU任务执行

GPU任务执行

提交后的效果如上，平台会设置将launcher 尽量调度到CPU机器，如果没有CPU机器则调度到GPU的机器，同时只分配到CPU的资源。

基于PS架构的分布式模型训练

虽然基于Ring Allreduce的模式在训练的性能方面会比PS架构要好很多，但是上面我也有提到过在推荐、广告、搜索等这种超大规模场景及需要做在线实时训练场景下PS架构是很适应的，所以在机器学习平台对这种分布式训练场景的支持是非常有必要的。

Multi-machine multi-GPU distributed parallel training

PS架构下所有的训练参数信息保存在参数服务上，参数服务是集群进行部署的，这样的话在超大规模参数下单机的内存资源是无法满足训练的要求的，特别像是在一些广告场景中，大量的Embedding造成参数规模很大。

PS架构的实现是基于Kubeflow的TrainingOperator来实现的，在TensorFlow的PS训练模式下：

从上面的图我们可以看到整个训练过程中会创建如下几种角色:

ps: 所有的参数存储的地方。

worker: 根据训练参数计算出梯度值，然后把梯度传递给ps之后拿到了ps返回的最新参数并更新到本地并进行多轮的迭代计算。

chief: 一般来说可以用来单独保存模型、代码执行点（比如执行构建Graph）、日志记录等等。比如部分代码只会在chief上运行。

这样我们就可以推断出TrainingOperator需要做的事情如下：

1: 创建ps/worker/chief等角色的容器

2: 根据这些角色创建的Pod的IP信息创建TensorFlow_CONFIG

3: 在创建容器时候设置TensorFlow_CONFIG为容器的环境变量

4: 在容器启动时执行用户脚本

TrainingOperator的整个处理流程并不是太复杂，对于机器学习平台来说就是创建TrainingOperator 对应的CRD：

1: 构建文件挂载信息，将分布式存储挂载到Horovod的容器中去，以保证在任何容器中能访问到训练脚本代码和配置、训练数据等等

2: 设置PS及chief容器的信息，这2种容器只需要分配到CPU的机器上即可，对于worker容器根据用户设置，如果需要CPU则设置CPU资源信息，如果需要GPU则设置GPU资源需求

3: 设置Node结点亲和度信息，如当前项目组下有资源，将当前任务的容器设置要调到到当前分组的资源结点下

4: 设置Pod之间的亲和策略

对于使用者来说只需要通过如下简单配置加上代码中的训练脚本配合就能就行分布式的训练了，对于资源的创建、回收，网络的管理都交由平台来管理，用户只需要关注自已的训练逻辑就可以了。

资源调度

在分布式计算下，我们需要申请大批量的机器进行训练，但是在大部的场景情况，无论是MPI+Horovod 或者是TensorFlow PS架构下都是需要等容器创建完后整个训练过程才会开始。

如上图，有一部分的worker申请到了机器了，但是另外几个worker申请不到机器，还一直处于Pending状态。

这里我们查看launcher的状态还一直处理init状态，等待所有的worker 准备好了后才开始作业，这些如果一直申请不到机器，已经起来的worker的资源就一直占用并浪费掉了，特别是GPU的资源，所以我们就需要一套资源调度框架来处理这些事情，原官方有kube-batch但是kube-batch已经很多年不更新了，对于目前很多的计算框架或者一些组件会有不兼容，volcano是当前行业中比较完善的调度框架。