分布式机器学习: 通信机制
文章目录
分布式机器学习与单机版的机器学习最大区别在于，它利用了多个工作节点同时训练、相互合作来加速学习过程。既然需要相互合作，那么通信就成为必不可少的环节。不过，分布式系统中的网络传输速度往往受限，导致通信常常成为分布式系统的瓶颈。举一个简单的例子：如果某个任务中计算与通信的时间占比为 1:1, 那么根据阿姆达尔定律（Amdahl's law），无论使用多少台机器做并行运算，其加速比都不会超过两倍。因此，分布式机器学习的关键是设计通信机制，从而降低通信与计算的时间比例，更加高效地训练出高精度的模型。
一、通信的内容
通信的内容与并行方式有关。但是无论是数据并行还是模型并行，都需要在各个工作节点之间进行相互通信。总体而言，通信的内容可以分为参数（或参数的更新）和计算的中间结果两类。
1.1 参数或参数的更新
在基于数据并行的分布式机器学习中，工作节点各自完成本地的学习任务，然后互相交流各自对模型的修改，或者直接同步各自的模型。因此，在此情形下通信的内容是模型的参数或者参数的更新。在很多机器学习任务中，参数以及参数的更新是稀疏的同时在训练过程中，随着模型趋于收敛，参数的更新也会越来越少，这都会使得通信量相对较少（或越变越少）。因此进行通信以获取参数和参数更新是一个比较高效的选择。
1.2 计算中间的结果
在基于模型并行的分布式机器学习中，通信内容往往是计算的中间结果。模型并行将一个完整的模型切分成若干小份，让每个工作节点负责其中一部分，共同协作来完成模型的训练。因为各个工作节点所负责的模型参数没有重叠，所以不需要进行通信以获取模型参数。然而，为了完成并行训练，不同的工作节点之间需要进行通信以获取相互依赖的中间计算结果。
具体而言，在前向传播时，数据从底层进入模型，沿着神经元之间的连边进行传播，从而产生中间层节点的激活函数值；在后向传播时，总体误差会从输出节点反向传播，从而产生中间层节点的误差信息和梯度更新值。这个过程中存在一些边连接着两个属于不同工作节点的子模型，于是我们需要按照连接关系在对应的工作节点之间进行通信，以供其完成各自的计算。
1.3 讨论
通信的目的是让工作节点互相交流它们各自的学习进展，不论是模型参数本身还是训练过程中的中间结果，本质上都是对于各个工作节点所获得的学习进展的表达方法。通过相互通信使总体的模型向正确的方向更新。
二、通信的拓扑结构
通信的拓扑结构指分布式机器学习系统中各个工作节点之间的连接方式。拓扑结构一般分为物理拓扑结构和逻辑拓扑结构两种。
早期由于数据量不大、模型不复杂时，分布式机器学习常常利用已有的分布式计算框架来实现通信，如消息通信接口（MPI）或者 MapReduce 计算框架。但也有本身的局限性，例如，使用 MPI 的方式，各个节点之间仅支持同步计算。
随着数据量増大，模型变得越来越复杂，人们设计了参数服务器这样的分布式机器学习系统。通过采用二部图的网络拓扑结构，参数服务器可以支持基于异步通信的并行训练。后来，随着深度学习的普及，机器学习系统将计算和通信统一抽象为一个数据图模型，通信可以在任意两个相连的图节点之间产生。
2.1 基于迭代式 MapReduce/ AllReduce 的通信拓扑
当我们把 MapReduce 的概念应用到分布式机器学习中，Map 操作定义了数据分发以及在本地工作节点上的计算，而 Reduce 操作则定义了全局参数的聚合过程。利用迭代式 MapReduce (IMR）操作可以实现典型的数据并行模式下的同步分布式机器学习算法。
另一种常用的分布式计算框架是消息通信接口（MPI）。程序设计人员主要使用其中的 AllReduce 接口来同步任何想要同步的信息，该接口支持所有符合 Reduce 规则的运算（比如求和、求平均、求最大值、求最小值等）。
分布式机器学习中基本的模型聚合方法主要是加和与平均，所以正好适合用 AllReduce 逻辑来处理。AllReduce 定义了一个标准接口，可以有多种实现方式。这些实现方式对应于不同的通信拓扑结构，包括星形拓扑、树形拓扑、蝶形拓扑等。各种拓扑结构在传输次数和传输量方面不尽相同，请参见下表。我们可以依据工作节点数和传输数据量选择合适的通信拓扑结构。
总体来说，不论是 IMR 还是 AllReduce 的模式都只能支持同步通信，并且从接口调用上可以看出，各个工作节点使用的逻辑都是统一的，同步时各个工作节点提供的信息都必须是针对同一组参数的。这也就暗示着要求每个工作节点能够处理完整的模型，这点对于模型规模很大的问题不太适用。
2.2 基于参数服务器的通信拓扑
采用 IMR 或 Allreduce 机制的分布式系统的训练速度往往取决于系统中最慢的节点；而更加严重的情况是，如果系统中有的工作节点不响应了（比如硬盘出现故障或者网卡出现问题），那么整个系统只能停下来，无法继续工作。其次，当机器学习任务中的模型参数非常多，已经超出了单个机器的内存容限时，IMR 或 AllReduce 架构也将无法胜任。
最后，近些年来研究人员提出了很多异步算法，这些算法无法由同步计算的框架实现。为了解决这些挑战，一种新型的分布式机器学习框架应运而生，那就是基于参数服务器的框架。
在参数服务器框架（上图) 中，系统中的所有节点被逻辑上分为工作节点（worker）和服务器节点（server）。各个工作节点主要负责处理本地的训练任务，并通过客户端接口与参数服务器通信，从参数服务器处获取最新的模型参数，抑或将本地训练产生的模型（或模型更新）发送到参数服务器。参数服务器框架中的灵魂是参数服务器（Parameter Server, PS）本身。PS 可以由单个服务器担任，也可以由一组服务器共同担任。可以看出，在逻辑上，参数服务器框架采用了二部图的通信拓扑结构。其中，工作节点和服务器节点之间彼此通信，而工作节点内部则无须通信。当服务器仅有一合时，便退化成为一个星形拓扑结构。
2.3 基于数据流的通信拓扑
在前面介绍的几种通信拓扑中，各个工作节点的运行逻辑是基本一致的，因此比较适合基于数据并行的分布式机器学习。
当我们进行基于模型并行的分布式机器学习时，则需要把不同类型的计算（例如不同子模型的更新）放置在不同的工作节点上。
近些年来，人们设计了基于数据流的分布式机器学习系统。在这种系统中，计算被描述为一个有向无环的数据流图。图中的每个节点进行数据处理或者计算，每条边代表数据的流动。当两个节点位于两台不同的机器上时，它们之间便会进行通信。
下图给出了一个工作节点逻辑的示例，每个节点实际上有两个通信通道：控制消息流和计算数据流。其中，计算数据流主要负责接收模型训练时所需要的数据、模型参数等，再经过工作节点内部的计算单元，产生输出数据（这里的数据可以是中间计算结果，也可以是参数更新），按需提供给下游的工作节点。控制消息流决定了工作节点应该接收什么数据，接收的数据是否已经完整，自己所要做的计算是否完成，是否可以让下游节点继续计算等。在工作节点定义时，需要指定工作节点的状态转换流程，从而在需要的时候生成一些信息，通过控制消息流通知后续节点准备进入消息接收和计算的状态。
其实数据流图是个很宽泛的概念，Mapreduce 和参数服务器的流程也可以用数据流图来表达。
2.4 讨论
在不同的分布式机器学习系统背后有着不同的通信拓扑结构，这些结构是研究人员和工程技术人员多年经验的积累，并且在实践中被大量使用。比如 MapReduce/AllReduce 在 Hadoop/Spark/REEF 中被大量使用，参数服务器被现在众多的大規模分布式机器学习系统（如 MxNet、Paddle、DMTK、Petumm）使用，而数据流图则被分布式深度学习框架所使用（如 TensorFlow）。它们各自存在于特定的应用场景，但分布式的思想却可以相互借鉴，因此将长期共存和共同发展，并推动分布式机器学习算法的不断创新。
三、通信的步调
所有的工作节点以同样的步调进行训练，这种通信模式称为同步通信。同步的通信步调能够保证分布式算法与单机算法的等价性，从而利于算法的分析和调试。但这需要各个工作节点之间彼此等待，造成计算资源闲置。因而这种方式具有算法上的优势但有系统上的劣势。
另一种方式则对所有工作节点的步调是否一致没有任何要求，称为异步通信。采用异步通信的方式时，各个机器可以按照自己的步调训练，无须彼此等待，从而最大化计算资源的利用率。但这种方式会使得各个工作节点之间的模型彼此不一致，存在延迟的问题。因而这种方式具有系统上的优势，但有算法上的劣势。在这两种极端的通信步调中间，还存在着一种折中的方式，以平衡同步和异步的优缺点。
3.1 同步通信
同步通信是指当集群中的一个工作节点完成本轮迭代后，需要等待集群中的其他工作节点都完成各自的任务，才能共同进行下一轮送代。
使用同步通信方式，可以确保各个工作节点模型的一致性。有些利用同步通信方式进行并行训练的分布式机器学习算法与其对应的单机优化算法等价。
但另一方面，同步方式由于要求各个机器之间的步调完全一致，会遇到掉队者（straggler）的麻烦。整个系统的效率取决于集群中运行最慢的节点。当参与分布式学习的工作节点之间存在显著性能差异时，同步通信很容易导致比较快的工作节点等待其他节点的现象。这个问题随着机器数量的増加变得愈加严重。因此，为了缓解这个问题，人们转而研究异步通信。
3.2 异步通信
异步通信是指当集群中的一个工作节点完成本轮迭代后，无须等待集群中的其他工作节点，就可以继续进行后续训练，因此系统效率可以大大提高（如下图所示）。然而，它会使得来自不同工作节点的模型参数之间存在延迟的现象，给模型聚合带来一定的挑战。
3.2.1 多机的异步通信
在多机异步通信系统中存在两种逻辑角色：本地工作节点（worker）和参数服务器节点（parameter server）。在学习的过程中，每个工作节点基于本地样本计算出参数更新（例如梯度），而参数服务器节点则负责保存和管理全局参数。在这样的框架中，各个工作节点之间是不需要相互通信的，因此它们可以完全按照自己的速度进行本地的模型训练，当完成一次本地的参数更新之后，直接通过参数服务器的 API，将更新推送到全局模型，随后就可以毫无顾忌地继续进行本地的下一轮参数更新。
但是会存在“延时”问题，举个例子，某个工作节点速度很快，它已经在全局模型的基础上往前训练了 100 轮；而另外一个工作节点速度慢，它オ在同一个全局模型的基础上往前训练了 1 轮。这很可能会减慢全局模型的收敛速率。
3.2.2 多线程的异步通信
当数据规模不太大时，大家通常会选择利用单机的多线程并行处理能力，而不是借助计算机集群来实现分布式计算。由于内存访问的速度远超过网络传输的速度，因而在规模不大时，这样并行所需要的时间更少。
在这类单机多线程并行的学习过程中，有多个线程同时访问模型参数，原则上需要对参数加锁来控制多线程访问中的冲突问题。然而，由于参数的更新速度很快，锁的获取所花费的时间在此类机器学习任务中是非常可观的，这往往会导致多线程的并行学习得不到理想的加速比。为了解決这个问题，人们提出了 Hogwild！算法。在 Hogwild！算法中，各个工作线程都直接无锁（lock-free）地读取和写人最新的模型及其更新。可以证明，在优化目标为凸函数且模型更新比较稀疏的情况下，异步无锁的写人不会对收敛性造成本质影响。因此，我们可以比较放心地使用多线程异步通信来实现快速的单机并行训练。在后续章节中我们还会专门介绍有关 Hogwild！算法的详细流程。
但是随着训练数据的继续增加，单机的计算能力还是会逊色于多机集群，因此大多数并行框架是工作在多机集群环境下的。实践中，通常会结合单机共享内存的本地加速方法和多机同步或异步的分布式机制共同完成大规模的机器学习任务。
3.3 同步和异步的平衡
下面我们将介绍其中一种比较经典的方法：延时同步并行（Stale Synchronous Parallel, SSP）
在极端情形下，异步通信可能存在非常大的延迟，从而导致学习过程收敛缓慢。但是实际系统中，我们通常遇到的情形又如何呢？答案是视集群的具体情况而论，也视实际的并行节点数目而论。
在实际的应用中，我们往往会采用一个相对同质化的集群（各个机器的计算性能和网络性能都趋于相同，并且相对稳定），并且不是所有时候都会有非常大量的节点参与运算。这时各个节点之间不存在非常明显的速度差异，偶尔有的机器快一点，有的机器慢一点，但是这种快慢变化大都是随机的，从相当长的一段时间来看，各个工作节点的平均速度应该趋于相同。在这种情况下，如果让各个工作节点异步执行，并且加上一定的控制逻辑，可能就不会出现之前那种令人担忧的情形了。
SSP 正是针对这种场景设计出来的。它的核心思想是控制最快和最慢节点之间相差的迭代次数不超过预设的國值。
下图对 SSP 的流程给出了形象的描述。在图中，阈值设为 3, 工作节点 1 是其中运算比较快的工作节点，而工作节点 2 是运算比较慢的工作节点。在工作节点 1 完成第 6 次更新的时候，工作节点 2 还在进行第 3 次更新。这时工作节点 1 已经领先太多，或者反过来说工作节点 2 所进行的更新的延迟太大了。这将会触发 SSP 算法中的等待机制。也就是说，此时工作节点 1 的最新参数请求将会被挂起，直到工作节点 2 到达第 4 次迭代位置才会解冻。在 SSP 的逻辑控制下，只要各个工作节点的迭代次数的差不超过预设的國值，则各个节点的运算就可以独立进行，不互相干扰。但是一且迭代次数差异太大就会触发一些等待，避免产生过大的延迟。
3.4 讨论
本节中我们讨论了不同的通信步调。不同的步调各有优缺点。在实践中，通常需要根据训练任务、数据规模、集群规模、使用场景等选择采用哪种方式。现阶段，很多工业应用中仍然在使用同步算法。这主要是由于同步算法的稳定性和可重复性很强，对实现产品的质量控制很有帮助。虽然同步算法的效率不高，但是也可以通过某种方式（如备份工作节点等）来提高速度。
四、通信的频率
在分布式机器学习中，通信是必要环节，同时也是相比于单机学习而言多出来的系统开销。通信与计算的时间比例往往决定了分布式机器学习系统加速比的上限。
本节我们讨论通信的频率。在设计分布式机器学习系统时，研究人员采用了多种方法来降低通信代价。这些方法主要利用机器学习算法的容错性特点，适当降低通信的频率，从而减少通信开销。在本章中，我们把通信频率分为时间频率和空间频率两种。其中，时间频率主要指通信的频次间隔，而空间频率主要指通信的内容大小。相应地，优化通信频率可以从两方面进行，即时域滤波和空域滤波。
4.1 时域滤波
时域滤波的方法旨在从通信的过程出发，控制通信的时机，减少通信次数，从而减少通信代价。采用时域滤波的主要方法有增加通信间隔、非对称的推送和获取以及计算和传输流水线。
4.2 空域滤波
空域滤波的方法旨在从通信的内容出发，尽量减少要通信的数据量，对传输的内容进行过滤、压缩或者量化，减少每一次传输所需的时间。接下来我们就介绍几种比较有代表性的做法。
4.2.1 模型过滤
种比较直观的方法是对模型参数进行过滤。其基本思想是，如果一次送代中某些参数没有明显变化，便可以将其过滤掉，从而减少通信量。
实践中发现，在模型训练的后期，通过这样的方法甚至可以过滤掉 99%的参数，而模型仍然可以收敛到原有的精度。
4.2.2 模型低秩化处理
模型过滤通过去除不重要的参数来减少通信量，而模型低秩化处理则通过低秩分解压缩参数来减少通信量。这种方法探索了参数中的低秩结构，其具体做法是通过矩阵低秩分解，如 SVD 分解，将原来比较大的参数矩阵分解成几个比较小的矩阵的乘积。在网络通信的时候实际传输的是分解得到的比较小的矩阵，在传输之后再重新恢复成比较大的参数矩阵。
需要注意的是：采用以上方法，虽然可以减少通信量，但是会带来额外的压缩与解压缩的开销。因而在实践中，通常需要权衡这些额外开销与减少的通信开销，从而得到更好的系统性能。
4.2.3 模型量化
除了参数过滤和参数矩阵低秩化处理之外，还有一类方法通过对要传输的信息的精度进行控制来降低通信代价。这种方法通过降低参数的每一维浮点数的精度来减少通信量。比如，在一比特量化方法中，将原本需要通过网络传输的参数更新信息，从 32 比特的浮点数压缩到了 1 比特的二进制数，从而把网络通信量压缩了 32 倍。具体的算法如下所示。
4.3 讨论
通信的效率对分布式机器学习的加速效果有很大影响，通常是分布式机器学习中的瓶颈。本节介绍了通信的频率，以及如何通过滤波的方式降低通信的频率。我们可以从时间上进行滤波，从通信的过程出发，减少全局通信的次数和时间。我们也可以在空间上进行滤波，从通信的内容出发，减少通信量和通信时间。我们讨论了模型过滤、低秩压缩以及模型量化三种空域滤波方法。实践中，这些方法也常常结合在一起使用，以获取最大限度的效率提升。
五、参考书籍
《分布式机器学习：算法、理论与实践》