基于 Horovod 进行深度学习分布式训练

原创李晨旭之家技术

总篇132篇 2022年第7篇

分布式训练概述

近年来深度学习在诸多领域得到广泛应用,深度学习模型良好的效果所依赖的大规模数据和模型，对训练方式和速度提出不断的挑战。为此，分布式训练的加速方法被越来越多地应用在深度学习领域。分布式训练采用多台 GPU/CPU 服务器, 通过构建高性能通信网络进行数据分发和模型同步，形成分布式深度学习计算的模式。分布式训练有数据并行和模型并行两种方式。

数据并行

模型规模不大，在一张 GPU 可以容纳，但是训练数据量会比较大，这时候就采用数据并行机制。数据并行适用于绝大部分深度学习模型训练加速。

数据并行将数据集均等地分配到各个计算节点，其中每个节点都有深度模型的一个副本及其权重，每个节点都会单独基于分配到的数据子集进行参数更新，然后再将每个结点上的参数进行聚合平均，计算得出一个新的全局参数权重。以此方式通过多次迭代得到最终的模型参数。通过最新的优化技术，数据并行能够在成千上万的GPU设备上训练非常大的数据集。

模型并行

当模型非常大，一张GPU已经存不下的时候，可以使用模型并行。模型并行是将模型的部分层或者单层拆分到多个GPU上并发计算，每个GPU执行指定的一部分计算任务，计算完成后通过通信操作完成模型计算结果的同步。对于一些超大型规模参数的模型，单个GPU无法存储整个模型计算过程中的数据，模型并行将模型的参数分配到不同的节点上进行计算，可以降低对结点内存的需求。由于每一层的计算依赖于前一层所有输出和当前层的所有参数，因此模型并行会在切分层后的每一步计算进行信息交换，通信负载较重。

常见的深度学习框架 TensorFlow、PyTorch、Apache MXNet 都提供内置方法以支持多GPU、多工作节点的分布式训练。除此之外，还有另外一种实现方法，即直接使用分布式深度学习框架，例如 Horovod 。

Horovod 介绍

Horovod 分布式训练

Horovod 是 Uber 公司打造的分布式深度学习开源框架，能够与TensorFlow、Keras、PyTorch 以及 Apache MXNet 等热门深度学习工具包协同使用。Horovod 使用 all-reduce 算法取代以往的参数服务器方法进行快速分布式训练，还提供了张量融合、梯度压缩、支持 NCCL通信等多种优化方法以进一步加快分布式训练的执行速度。

Horovod目的是快速的实现分布式深度学习，只需修改几行 Python 代码，就能够快速、轻松地训练现有的训练脚本，使其可在数百个 GPU 上运行,将模型训练时间从几周和几天缩短到几小时和几分钟。Horovod 以其易用性和高性能，在业界得到了广泛应用。

Horovod 还可以运行在 Apache Spark 之上，允许在单个 Pipeline 下统一数据处理、模型训练和模型验证。只要集群中配置了 Horovod，就可以方便的分布式训练基于TensorFlow、PyTorch、MXNet等框架的深度模型。

Horovod 主要是支持数据并行的深度学习分布式训练。

Horovod性能

Horovod 官方使用TensorFlow 基准测试和在此基础上修改的Horovod实现，对比了Inception V3 和 ResNet-101两个模型在不同数量的GPU卡上图像处理的速度。

通过对比，可以看到Horovod的扩展能力有了很大的提高。在GPU卡数越多时，提升越明显。在128卡上，同时使用 Inception V3 和 ResNet-101 模型实现了 88% 的效率提升，训练速度大约是标准TensorFlow 分布式的两倍。

下图是我们在生产环境中使用V100卡，对比TensorFlow 基准测试和在此基础上修改的Horovod实现，使用Resnet50模型测试对比的结果，可以看到在单机多卡时，性能并没有明显差异，但是在多机多卡时，Horovod 有明显的效果提升。

Horovod 架构

Horovod主要由数据通信层、通信控制层、深度学习框架接口层、启动层四部分组成。其中启动层通过horovodrun或mpirun启动训练进程，之后每个训练进程通过调用TensorFLow、PyTorch、MXNet等框架进行单个结点的数据输入、参数更新，在每个进程完成一个或多个batch计算后，得到的Tensor（参数）通过MPI或GLoo控制进行ring-allreduce，ring-allreduce 的通信可以基于MPI、NCLL、DDL、MLSL或GLoo。

Horovod 通信机制

2017 年初，百度发表了ring-allreduce技术，用于GPU间梯度同步平均。通过将GPU卡的通信模式拼接成一个环形，每个 GPU 只从左邻居接受数据、并发送数据给右邻居。从而减少随着卡数增加而带来的资源消耗。意识到ring-allreduce的优点后，为能更广泛的使用ring-allreduce,因此开发了Horovod。Horovod在GPU上使用NCCL的ring-allreduce，在CPU上使用MPI、GLoo或oneCCL实现ring-allreduce。

Ring-allreduce算法主要分两步：

1. scatter-reduce：逐步交换彼此的梯度并融合，最后每个 GPU 都会包含完整融合梯度的一部分

2. allgather：GPU 会逐步交换彼此不完整的融合梯度，最后所有 GPU 都会得到完整的融合梯度。

举例：数组求和

scatter-reduce：将数组在每个GPU上都分块——>N-1轮的scatter-reduce，每一轮中，每个GPU将自己的一个chunk发给右邻居，并接收左邻居发来的chunk，并累加。

Allgather:和scatter-reduce操作类似，只不过将每个chunk里面的操作由累加值变为替换。

通信代价分析：每个 GPU在Scatter Reduce阶段，接收N-1次数据，N是GPU数量；每个GPU在allgather阶段，接收 N-1次数据；每个GPU每次发送K/N大小数据块，K 是总数据大小；所以，Data Transferred=2(N−1)*K/N ，随着GPU数量N增加，总传输量恒定。也就是理论上，随着gpu数量的增加，ring all-reduce有线性加速能力。

Horovod 训练加速

信息压缩：完整精度会使用与本地模型相同的 32 位浮点数（float32）进行传输，模型较大时通信会成为瓶颈，信息压缩通过量化、降低精度或稀疏化等方法压缩梯度，再用压缩后的梯度更新参数。在很多场景下，可以达到和完整精度相同的效果，同时提升通信效率。Horovod在allreduce 时支持使用压缩算法，以减少每个参数更新步骤期间发送的数据量。

Tensor Fusion：梯度传递的时候可以将小的tensor合并成一个大的tensor再进行传递，从而减小每一次操作的额外开销(overhead)。那些具有大量tensor的模型，例如 ResNet-101，往往有很多微小 tensor的 allreduce操作。而 ring-allreduce 在张量足够大时能以最佳效率利用网络，如果tensor非常小，则无法有效或快速地通信。为了整合这些小的tensor, Horovod 团队发明了 Tensor Fusion，在调用 Horovod 的 ring-allreduce 之前将 tensor 融合在一起。Tensor Fusion 在未优化的 TCP 网络上针对大量层的模型性能提高了约 65%。

horovod on k8s

代码修改

Horovod是基于MPI模式进行设计实现的，主要有size,rank,local_rank, allreduce,allgather,broadcast,and alltoall这些核心概念。例如有两台两卡的GPU机器，为每一个GPU启动一个训练进程，其中：

size是总的进程数，这里是4。
rank是所有进程的唯一进程ID，值为0-3（size-1）。
local_rank是单个GPU机器上的唯一进程ID，值为0-1。
Allreduce是将多个进程上的数据聚合再将聚合结果分发给所有进程的操作，Allreduce用于平均张量。
Allgather是在一个进程中收集所有进程数据的操作，Allgather用于收集稀疏张量。
broadcast是从一个进程广播数据到所有其他进程的操作。
Alltoall是一种在所有进程之间交换数据的操作, Alltoall有助于模型并行。

代码修改示例如下：

1. 初始化 horovod

import horovod.tensorflow.keras as hvdhvd.init()

2.为每个 worker 分配 GPU

config = tf.ConfigProto()config.gpu_options.visible_device_list = str(hvd.local_rank())K.set_session(tf.Session(config=config))

3. 调整学习率和分布式优化器

opt = keras.optimizers.Adadelta(lr=0.01 * hvd.size())   #推荐使用LR N = LR 1 * Nopt = hvd.DistributedOptimizer(opt)#推荐

4. worker同步初始化参数

callbacks = [hvd.callbacks.BroadcastGlobalVariablesCallback(0)]

5. 某个worker上保存Checkpoint

if hvd.rank() == 0:
callbacks.append(keras.callbacks.ModelCheckpoint('./checkpoint--{epoch}.
model.fit(x_train, y_train, callbacks)

6. 运行Horovod

单机多卡:$ horovodrun -np 4 -H localhost:4 python train.py多机多卡:$ horovodrun -np 16 -H server1:4,server2:4,server3:4,server4:4 python train.py

多机多卡运行时，需要打通多个结点间的无密SSH访问。

性能表现

推荐多目标模型Horovod多机多卡训练效果如下图：

图像领域基于inception V4模型的车系识别使用Horovod多机多卡训练效果如下图：

NLP领域基于Bert模型的标题生成使用Horovod多机多卡训练效果如下图：

三个实际生产中使用的模型通过Horovod进行分布式训练，可以看到随着GPU卡数的增加，训练时长在接近线性缩短。基本符合Horovod官方测试的效果。

VS Bagua

Bagua（八卦）是快手和苏黎世理工宣布开源分布式训练框架,和Horovod 类似，主要支持数据并行的分布式训练。Bagua和horovod都是在现有深度学习框架（tensorflow、pytorch等）的基础上进行分布式通信优化，以提升分布式训练的效率。bagua相比horovod有模型参数异步通信，更好的信息压缩算法，及与去中心化的组合，因此在性能上有一定的优势，但是在支持的深度学习框架、文档全面性、易用性、稳定性、生态活跃度、扩展性等方面全面，horovod要优于Bagua。在两台4卡V100、32G显存的GPU机器作了简单对比，对比数据如下：

在图像数据集上，相同的迭代次数，bagua平均每秒处理的样本数多5%-10%，训练用时短5%-9%，精度（acc）高约1%。在nlp数据集上，相同的迭代次数，bagua训练用时在2机4卡和2机2卡分别多7%、4%，在2机1卡短3%，但是f1分数分别高2%，在相同的时间内，bagua能取得更高的f1分数。

horovod on spark

功能介绍

Horovod on Spark是对 Horovod 的封装，可以使 Horovod 运行在 Spark 集群，使得在 Spark 集群中可以轻松运行深度学习分布式训练作业。Horovod on Spark 有如下优势：

1、如果训练数据源自 Spark ，则数据处理、模型训练和模型评估这一个机器学习的循环都放在Spark技术栈之中。

2、充分利用大规模基于 CPU 的 HDFS 集群，尤其是适合某些 GPU 计算量少而 CPU 计算量多的模型。

3、Horovod 需要手动实现数据分片，Horovod on spark 可以轻松通过 DataFrame 实现数据分片。

Horovod on spark 的机制是首先通过yarn调度分配资源，然后Horovod在分配好的Spark cluster上启动训练进程。Horovod on Spark 是在 Spark 之上实现了一套自己的 DriverService 和 TaskService，分别对应 Spark 的 Driver 和 Executor。当 Spark 的Driver 和 Exexutor 创建之后，Horovod DriverService接管进行脚本分发，Task创建和注册，可以在 M 个 Exexutor 上 N个 tasks, N 可以大小于、等于、小于M,一般设置为等于 M，不然会出现资源的抢占或空闲。

Horovod 支持的基于 TensorFlow、Keras、PyTorch 和 Apache MXNet 的分布式深度学习训练作业，都可以方便的迁移到 Spark 上进行训练。

Horovod on spark 提供了两种在 Spark 上使用 Horovod的 API：分别是高级的 Estimator API 和低级的 Run API。两者使用相同的底层机制在 Spark Exexutor上运行 Horovod。

Estimator API 使用方式和 PySpark Estimator 一致，抽象了数据处理、模型训练循环、模型保存、指标收集和分布式训练，使用Spark DataFrame做为输入数据源。缺点是现在的版本只支持 Keras (tf.keras或keras) 和 PyTorch。

Run API 是对 Horovod 脚本的直接调用，在 Kubernetes 运行的 Horovod 脚本可以不做修改，通过Run API简单封装后就可以在 Spark 上运行，Horovod 支持的深度学习框架都可以运行在 Spark 上，而且不需要了解太多 Spark 机制。

安装运行

Horovod on spark的安装有两种方式，一种是将 Horovod 安装在 Spark 集群的所有结点上，开启 Horovod on Spark 的支持。该方案的缺点是安装、维护、升级成本都非常高，对算法框架的个性化支持也非常困难。另一种比较好的方式构建 Python 虚拟环境，虚拟环境方式安装 Horovod、TensorFlow 等工具包，在提交作业时使用构建的虚拟环境。以上安装方式都默认使用GLOO进行通信，MPI 需要单独编译安装，并在submit时指定统一的 LD_LIBRARY_PATH 和 PKG_CONFIG_PATH环境变量，且运行时存在Spark重复初始化的问题，暂时在Spark上不使用MPI。

虚拟环境构建和 Horovod 安装方式可参考官方安装文档。在 Spark 作业Submit时通过spark.yarn.dist.archive指定需要使用的虚拟环境包，通过设置appMasterEnv和executorEnv的PYSPARK_PYTHON、PYSPARK_DRIVER_PYTHON指定虚拟环境包 Python 的路径。这样作业提交后，虚拟环境会分发到Driver和所有 Executor 上，并使用虚拟环境的 Python 执行提交的训练脚本。

使得虚拟环境包时需要修改 Horovod 源码解决以下两个问题：

1、Driver和Executor 上Python路径不一致问题。

2、Driver和Executor 上hdfs_tokens路径不一致问题。

为什么以上两上环境变量要在Driver和所有 Executor 上一致，这个是Horovod on spark 的机制决定的。Horovod on spark 通过 DriverService 分发执行命令和脚本在各个 Executor 上进行启动，在分发时 Python 和 hdfs_tokens 的绝对路径被分发到了各个 Executor 上，这样在 Executor 上执行启动命令时就会出错，找不到Python和 hdfs_tokens。有两种修改方式，一种是改为相对路径，一种是通过软链接设置统一的路径。这里以软链接为例，在DriverService 和 TaskService初始化时，在对应的 Executor上生成/tmp/horovod_spark_${user}/${appid}路径，再将当前Executor的执行目录软链接到新建的目录，这样保证了所有Driver和所有 Executor上Python和hdfs_tokens路径的一致性，之后再修改分发的命令，在gloo_run中修改sys.executable或executable为统一的软链接目录，同时修改env['HADOOP_TOKEN_FILE_LOCATION']为软链接中hdfs_tokens的路径。

Horovod 在 Spark上运行，还需要注意网络通信时interface的选取。Executor之间通信时会选择结点上所有interface 列表中的第一个，在某些网络中可能会取到 127.0.0.1或其他网络地址，导致通信不能连接，可以通过在get_local_addresses中过滤其他interface, 以保证取到的是局域网物理地址。

代码修改

Horovod estimator可以参考官方examples，Horovod run 通过如下方式封装后直接调用原Horovod训练代码，某些Horovod训练代码实现需要做一些简单修改，主要涉及到args的获取和函数入口的定义。

args = parser.parse_args()
conf = SparkConf()spark = SparkSession.builder \    .config(conf=conf) \    .getOrCreate()
def train_fn():    import sys    sys.path.append("./__pyfiles__")    import mnist_keras              #导入horovod训练脚本    return mnist_keras.main(None)  #启动horovod训练脚本
import horovod.spark
horovod.spark.run(train_fn, args=(), num_proc=args.num_proc, verbose=2, prefix_output_with_timestamp=True )
spark.stop()

VS Xlearning，TensorFlowon spark

XLearning是奇虎360公司发布的深度学习调度平台，XLearning平台将大数据与深度学习相融合，基于Hadoop Yarn完成了对TensorFlow、MXNet、PyTorch、Keras等常用深度学习框架的集成，是典型的“AI on Hadoop”的实现。

与Horovod on spark相比XLearning是深度学习调度平台，主要是通过Hadoop yarn调度的方式，将TensorFlow、MXNet、PyTorch等深度学习训练进程跑在Hadoop集群，支持这些深度学习框架自带的分布式训练机制。

Horovod on spark是深度学习分布式训练平台，通过spark实现了Hadoop yarn调度和资源分配，之后的分布式训练完全由Horovod实现，通过张量融合、梯度压缩、NCCL通信等多种优化方法对分布式训练进行加速。另外Horovod on spark还基于spark estimator实现了自己的estimator，实现了数据处理和模型训练的pipeline统一。

TensorFlow on spark是yahoo开源的基于spark进行分布式tensorflow训练的开发框架。与Horovod on spark相似，也利用spark编程API的方式启动多个worker到yarn集群中运行，以此解决了计算资源分配问题。对于每个worker，基于TF_CONFIG环境变量启动tensorflow训练脚本，之后就是tensorflow本身的分布式训练机制进行训练。和XLearning相似，本质上是一个深度学习调度平台，并没有对分布式训练进行优化回事。

总结

本文主要阐述了Horovod分布式深度学习训练的一些机制和实现，以及实际生产环境中应用的情况。Horovod 易用且高效，统一了分布式训练机制，不用分别去学习各个深度学习框架各自的分布式训练机制，大大降低了学习和使用成本。同时，Horovod更新也非常快，不断带来更多新的功能，比如分布式训练自动寻参、模型并行等，助力分布式深度学习训练更便捷，更高效。

参考资料：

1. Sergeev, A., Del Balso, M. (2017) Meet Horovod: Uber’s Open Source Distributed Deep Learning Framework for TensorFlow. Retrieved from https://eng.uber.com/horovod/

2. Sergeev, A. (2017) Horovod - Distributed TensorFlow Made Easy. Retrieved from https://www.slideshare.net/AlexanderSergeev4 /horovod-distributed-tensorflow-made-easy

3. Sergeev, A., Del Balso, M. (2018) Horovod: fast and easy distributed deep learning in TensorFlow.Retrieved from arXiv:180 2.05799

4. JOOST VERBRAEKEN, MATTHIJS WOLTING, A Survey on Distributed Machine Learning, Retrieved from https://arxiv.org/ pdf/1912.09789

5. Travis Addair,Gitansh Chadha, Distributed deep learning with Horovod, Retrieved from https://d1.awsstatic.com/events/reinvent /2019/Distributed_deep_learning_with_Horovod_AIM418.pdf

6. 朱泓睿，元国军，姚成吉，谭光明，王展，户忠哲，张晓扬，安学军，分布式深度学习训练网络综述，https://crad.ict.ac.cn/CN/article/ downloadArticleFile.do?attachType=PDF&id=4332

作者简介

汽车之家

李晨旭

智能数据中心

负责汽车之家机器学习平台分布式训练的支持，帮助算法团队实现简单、高效的分布式训练，快速迭代模型。

阅读更多

▼ 关注「之家技术」，获取更多技术干货 ▼

继续滑动看下一个