版权声明：本文为博主原创文章，未经博主允许不得转载。 https://blog.csdn.net/muumian123/article/details/86374058
分布式Tensorflow底层的通信是gRPC。gRPC首先是一个RPC，即远程过程调用，通俗的解释是：假设你在本机上执行一段代码num=add(a,b)，它调用了一个过程call，然后返回了一个值num，你感觉这段代码只是在本机上执行的，但实际情况是，本机上的add方法是将参数打包发送给服务器，然后服务器运行服务器端的add方法，返回的结果再将数据打包返回给客户端。
TensorFlow集群就是一组任务，每个任务就是一个服务。服务由两个部分组成，第一部分是master，用于创建session，第二部分是worker，用于执行具体的计算。TensorFlow一般将任务分为两类job：一类叫参数服务器，parameter server，简称为ps，用于存储tf.Variable；一类就是普通任务，称为worker，用于执行具体的计算。
一般而言，机器学习的参数训练过程可以划分为两个类别：第一个是根据参数算算梯度，第二个是根据梯度更新参数。对于小规模训练，数据量不大，参数数量不多，一个CPU就足够了，两类任务都交给一个CPU来做。对于普通的中等规模的训练，数据量比较大，参数数量不多，计算梯度的任务负荷较重，参数更新的任务负荷较轻，所以将第一类任务交给若干个CPU或GPU去做，第二类任务交给一个CPU即可。对于超大规模的训练，数据量大、参数多，不仅计算梯度的任务要部署到多个CPU或GPU上，而且更新参数的任务也要部署到多个CPU。如果计算量足够大，一台机器能搭载的CPU和GPU数量有限，就需要多台机器来进行计算能力的扩展了。参数服务器是一套分布式存储，用于保存参数，并提供参数更新的操作。
1 TensorFlow集群创建
每个任务用一个ip:port表示。TensorFlow用tf.train.ClusterSpec表示一个集群信息。
import tensorflow as tf
cluster = tf.train.ClusterSpec({
"worker": [
"worker_task0.example.com:2222",# /job:worker/task:0 运行的主机
"worker_task1.example.com:2222",# /job:worker/task:1 运行的主机
"worker_task2.example.com:2222" # /job:worker/task:2 运行的主机
],
"ps": [
"ps_task0.example.com:2222", # /job:ps/task:0 运行的主机
"ps_task1.example.com:2222" # /job:ps/task:1 运行的主机
]})
上面的语句提供了一个TensorFlow集群信息，集群有两类任务，称为job，一个job是ps，一个job是worker；ps由2个任务组成，worker由3个任务组成。
2tf.train.Server创建每个任务
tf.app.flags.DEFINE_string("job_name", "worker", "One of 'ps', 'worker'")
tf.app.flags.DEFINE_integer("task_index", 0, "Index of task within the job")
FLAGS = tf.app.flags.FLAGS
def main(_):
server = tf.train.Server(
cluster,
job_name=FLAGS.job_name,
task_index=FLAGS.task_index
)
server.join()
if __name__ == "__main__":
tf.app.run()
一个tf.train.Server包含了：本地设备(GPUs，CPUs)的集合,可以连接到其它task的ip:port(存储在cluster中)，还有一个session target用来执行分布操作。还有最重要的一点就是，它创建了一个服务器监听port端口，如果有数据传过来，他就会在本地执行(启动session target，调用本地设备执行运算)，然后结果返回给调用者。
3 指定分布式设备
with tf.device("/job:ps/task:0"):
weights_1 = tf.Variable(...)
biases_1 = tf.Variable(...)
with tf.device("/job:ps/task:1"):
weights_2 = tf.Variable(...)
biases_2 = tf.Variable(...)
with tf.device("/job:worker/task:0"): #映射到主机(10.1.1.1)上去执行
input, labels = ...
layer_1 = tf.nn.relu(tf.matmul(input, weights_1) + biases_1)
logits = tf.nn.relu(tf.matmul(layer_1, weights_2) + biases_2)
with tf.device("/job:worker/task:1"): #映射到主机(10.1.1.2)上去执行
input, labels = ...
layer_1 = tf.nn.relu(tf.matmul(input, weights_1) + biases_1)
logits = tf.nn.relu(tf.matmul(layer_1, weights_2) + biases_2)
# ...
train_op = ...
with tf.Session("grpc://10.1.1.2:3333") as sess: #在主机(10.1.1.2)上执行run
for _ in range(10000):
sess.run(train_op)
with tf.Session("grpc://..")是指定gprc://..为master，master将op分发给对应的task。
4同步更新和异步更新
同步更新指的是：各个用于并行计算的电脑，计算完各自的batch 后，求取梯度值，把梯度值统一送到ps服务机器中，由ps服务机器求取梯度平均值，更新ps服务器上的参数。如下图所示，可以看成有四台电脑，第一台电脑用于存储参数、共享参数、共享计算，可以简单的理解成内存、计算共享专用的区域，也就是ps job；另外三台电脑用于并行计算的，也就是worker task。
图1 同步更新
这种计算方法存在的缺陷是：每一轮的梯度更新，都要等到A、B、C三台电脑都计算完毕后，才能更新参数，也就是迭代更新速度取决与A、B、C三台中，最慢的那一台电脑。
异步更新指的是：ps服务器收到只要收到一台机器的梯度值，就直接进行参数更新，无需等待其它机器。
图2 异步更新
这种迭代方法比较不稳定，收敛曲线震动比较厉害，因为当A机器计算完更新了ps中的参数，可能B机器还是在用上一次迭代的旧版参数值。
5 图内模式和图间模式
in-graph模式下数据分发在一个节点上。这种方式配置简单，其他结算节点只需join操作，暴露一个网络接口，等在那里接受任务就好。但坏处就是训练数据的分发在一个节点上，要把训练数据分到不同的机器上，严重影响了并发的训练速度。
between-graph模式下，训练的参数保存在参数服务器，数据不用分发，数据分片的保存在各个计算节点，各个计算节点自己算自己的，算完后把要更新的参数告诉参数服务器，参数服务器更新参数。这种模式的优点是不用进行训练数据的分发，尤其数据量在TB级的时候，节省了大量的时间，所以大数据深度学习推荐使用between-graph模式。
6分布式TensorFlow与Spark对比：
- 分布式的级别不同：TensorFlow的Tensor、Variable和Op不是分布式的，分布式执行的是subgraph. Spark的op和变量都是构建在RDD上，RDD本身是分布式的。
- 异步训练：TensorFlow支持同步和异步的分布式训练；Spark原生的API只支持同步训练
- 分布式存储：Spark在底层封装好了worker和分布式数据之间的关系；TensorFlow需要自行维护。
- Parameter Server：TensorFlow支持，Spark暂不支持。
- TF分布式部署起来还是比较繁琐的，需要定义好每个任务的ip:port，手工启动每个task，不提供一个界面可以对集群进行维护。
7 TensorFlowOnSpark
上面介绍的是直接使用TensorFlow进行手工配置，通过修改部分代码即可实现分布式计算的过程。通过调研发现，Yahoo于2017年开源了基于Spark的Tensorflow，使用executor执行worker和ps task。TensorFlowOnSpark为Apache Hadoop和Apache Spark 集群带来了可扩展的深度学习。通过结合深度学习框架TensorFlow和大数据框架Apache Spark和Apache Hadoop的显着特性，TensorFlowOnSpark可在GPU和CPU服务器集群上实现分布式深度学习。
TensorFlowOnSpark在Apache Spark集群上启用分布式TensorFlow训练和推理。它旨在最大限度地减少在共享网格上运行现有TensorFlow程序所需的代码更改量。
TensorFlowOnSpark具有以下优势：
- 使用<10行代码更改轻松迁移所有现有TensorFlow程序;
- 支持所有TensorFlow功能：同步/异步训练，模型/数据并行，推理和TensorBoard;
- 服务器到服务器的直接通信在可用时实现更快的学习;
- 允许HDFS上的数据集和其他源由Spark推送或由TensorFlow拉取;
- 轻松集成现有的数据处理流水线和机器学习算法（例如MLlib，CaffeOnSpark）;
- 轻松部署在云端或内部部署：CPU和GPU，以太网和Infiniband。
TensorFlowOnSpark开源时间较短，未得到充分验证，并且安装配置过程十分艰难（网络资料显示，使用Yahoo提供的配置过程无法完成集群搭建任务，会遇到各种坑）。