【机器学习】并行计算（parallel computation）Part1

为什么我们在机器学习中需要用到并行计算呢，因为现在最流行的机器学习算法都是神经网络，神经网络模型的计算量、参数量都很大，比如ResNet-50参数量为25M。而我们在训练的时候使用的数据集也很大，比如ImageNet数据集含有14M张图片。用大数据来训练大模型会产生很大的计算成本，比如用一个NVIDIA M40 GPU在ImageNet数据集上训练ResNet-50需要耗费14天的时间，如果我们采用的是并行计算，便可以使用多个GPU进行计算加快钟表时间（wall-clock time），但并不会计算总GPU时间，因为总计算量并没有减少。

Parallel Gradient Descent 并行梯度下降

线性回归（linear regression）的输入是一个向量x，输出是 $f(x)=W^{T}X$ ，那我们如何确定W呢，就需要用到最小二乘法（Least squares regression）:

$W^{*}=\underset{W}{min}L(W)=\underset{W}{min}\sum_{i=1}^{n}\frac{1}{2}(w^{T}x_{i}-y_{i})^{2}$

在求最优解的时候，我们需要用到并行梯度下降法（Parallel Gradient Descent）。正常情况下，我们需要求出损失函数来，并计算它的梯度，然后利用学习率（步长stride）得到下一个梯度值。而并行梯度下降需要将梯度在不同处理器上进行计算，每块处理器只做了部分计算，相加之后就得到总梯度。

Communication 处理器之间的通信

做并行梯度下降要对数据作划分，需要将参数w和梯度g进行传递。因为要使用多个处理器，所以要考虑处理器之间的通信问题。有两种通信方式，一种是Share memory，另一种是Message passing。在Share memory中，一个处理器可以看到其他处理器得到的结果，但是共享内存的通信没有办法做到大规模的并行。

而Message passing有多个节点，每个节点都有多个处理器，每个节点内的处理器共享内存，但是节点1看不到节点2的内存。节点之间通信需要用到message passing，比如用到TCP、IP协议将共享文档打包为package。

那我们如何进行节点之间的协调呢，有两种方法—— Client-Server架构和Peer-to-Peer架构。Client-Server架构，把一个节点作为Server用来协调其他的节点，把其他的节点都作为Worker，用来做计算。

Peer-to-Peer架构，这种架构没有Server，所有节点都被拿来计算。每个节点都有几个邻居，邻居之间可以进行通信。

Synchronous Parallel Gradient Descent Using MapReduce 用MapReduce实现同步并行梯度下降

MapReduce是由Google开发的一个软件系统，用来做大规模的数据分析和机器学习。有系统设计，但无开源代码，后来大家把这种分布式的变成模型都叫做MapReduce。它的架构采用的是Client-Server，通信方式是Message-passing，并行是同步的bulk synchronous parallel，每一轮需要等到所有的worker全都完成工作，才能进行下一轮。

Apache Hadoop是MapReduce的开源实现，Apache Spark是MapReduce更好的开源实现，相比之前的将所有内容写入内存而非磁盘，有很好的容错机制，速度快很多。MapReduce编程模型很适合用来做大数据处理，但做机器学习并不是很高效。

MapReduce的架构Server和Worker之间可以进行通信，Server可以将信息广播到Worker节点上，这叫做Broadcast。比如作平行梯度下降的时候，Server需要将模型参数广播出去，每个Worker节点都可以做计算。

如果我们要实现算法，需要自己定义一个函数，然后所有的Worker都会运行这个函数，这一步叫做Map，Map操作是由所有Worker并行做的。做并行梯度下降的时候，每一个Worker都用自己的数据做部分计算。

Reduce操作也需要进行通信，Worker会把他们的计算结果传回Server，然后Server将结果进行整合（sum，collect，mean等）。

用MapReduce作并行梯度下降，数据并发（Data parallelism）意思是数据划分到Worker节点上，每个Worker都有部分数据样本，这个例子中，我们采用了m个节点，每个节点上存了3个数据样本。

用MapReduce实现并行梯度下降会有以下几个步骤：

BroadCast：Server首先将最新的参数 $w_{t}$ 广播到所有Worker节点上
Map：Worker作本地操作，将 $(x_{i},y_{i},w_{t})$ 映射为 $g_{i} = (w_{i}^{T}x_{i}-y_{i})x_{i}$ ，然后得到n个向量 $g_{1},g_{2},....,g_{n}$
Reduce：计算最终的梯度 $g=\sum_{i=1}^{n}g_{i}$ 。每个Worker计算存在本地内存中的所有 ${g_{i}}$ 得到一个向量，然后Server将m个向量进行求和。
Server更新参数： $w_{t+1}=w_{t}-\alpha g$

每个Worker只存储 $\frac{1}{m}$ 的数据，只进行 $\frac{1}{m}$ 的运算。因此，运行时间在理论上就会减少到 $\frac{1}{m}$ ，但是并行计算还有通信（communication）和同步（synchronization）的代价。如果将通信和同步时间算上，肯定不是 $\frac{1}{m}$ 。在做并行计算的时候，通常考虑speedup ratio，调整节点数量得到图像，理想情况下不考虑通信和同步代价的话m个节点的时钟时间与1个节点的时钟时间成正比。

通信代价由两部分组成，通信复杂度（Communication complexity）和 时延（Latency）。通信复杂度指的是多少个words或者是bits在Server和Worker之间传输，它与模型参数量成正比，并且会随着Worker节点数增多而增长。时延指的是数据包从一个节点发送到另一个节点所花费的时间，这是由计算网络决定的。总通信时间可以表示为 $\frac{complexity}{bandwidth}+lantency$ 。

而同步（Bulk Synchronous）代价是指的如果m个节点在做计算（Map），其中有一个节点计算很慢，那么其他节点则需要等待这一个节点完成计算；比如正常情况下节点的任务量应该是差不太多的，但是如果一个节点挂掉了则需要进行重启，这个节点就会比其他正常的节点慢很多，这个节点也被称为straggler，而随着节点数量的增多，出现straggler的可能性也会增大。Straggler effect指的是所有的Worker需要等最慢的Worker，时钟时间是取决于最慢的Worker，这是由同步问题产生的。