分布式TensorFlow

本文档介绍如何建立TensorFlow服务器集群，以及如何在该集群中分发计算图。咱们假设您熟悉编写TensorFlow程序的基本概念。

开始

要查看一个简单的TensorFlow集群，请执行如下操做：

# Start a TensorFlow server as a single-process "cluster".

$ python
>>> import tensorflow as tf
>>> c = tf.constant("Hello, distributed TensorFlow!")
>>> server = tf.train.Server.create_local_server()
>>> sess = tf.Session(server.target)  # Create a session on the server.
>>> sess.run(c)
'Hello, distributed TensorFlow!'

该tf.train.Server.create_local_server 方法建立具备进程内服务器的单进程集群。

建立一个集群

<iframe width="412" height="232" src="https://www.youtube.com/embed/la_M6bCV91M?ecver=1" frameborder="0" allowfullscreen></iframe>

TensorFlow“集群”是参与TensorFlow图形的分布式执行的一组“任务”。每一个任务与TensorFlow“服务器”相关联，该服务器包含可用于建立会话的“主”，以及在图中执行操做的“工做者”。群集还能够分为一个或多个“做业”，其中每一个做业包含一个或多个任务。

要建立集群，请在集群中为每一个任务启动一个TensorFlow服务器。每一个任务一般在不一样的机器上运行，但您能够在同一台机器上运行多个任务（例如控制不一样的GPU设备）。在每一个任务中，执行如下操做：

1. 建立一个tf.train.ClusterSpec描述集群中全部任务的内容。这对每一个任务应该是同样的。

2. 建立一个tf.train.Server，传递tf.train.ClusterSpec给构造函数，并使用做业名和任务索引识别本地任务。

建立一个tf.train.ClusterSpec描述集群

群集规范字典将做业名称映射到网络地址列表。将此字典传递给tf.train.ClusterSpec 构造函数。例如：

tf.train.ClusterSpec construction	Available tasks
tf.train.ClusterSpec({"local": ["localhost:2222", "localhost:2223"]})	/job:local/task:0 /job:local/task:1
tf.train.ClusterSpec({ "worker": [ "worker0.example.com:2222", "worker1.example.com:2222", "worker2.example.com:2222" ], "ps": [ "ps0.example.com:2222", "ps1.example.com:2222" ]})	/job:worker/task:0 /job:worker/task:1 /job:worker/task:2 /job:ps/task:0 /job:ps/task:1

tf.train.Server在每一个任务中建立一个实例

一个tf.train.Server对象包含了一组本地设备，一组到其余任务的链接 tf.train.ClusterSpec，而且 tf.Session可以使用这些来进行分布式计算。每一个服务器是特定命名做业的成员，而且在该做业中具备任务索引。服务器能够与群集中的任何其余服务器进行通讯。

例如，推出了集群两台服务器上运行localhost:2222 ，并localhost:2223在本地机器上的两个不一样的进程运行下面的代码片断：

# In task 0: cluster = tf.train.ClusterSpec({"local": ["localhost:2222", "localhost:2223"]}) server = tf.train.Server(cluster, job_name="local", task_index=0)   # In task 1: cluster = tf.train.ClusterSpec({"local": ["localhost:2222", "localhost:2223"]}) server = tf.train.Server(cluster, job_name="local", task_index=1)

注意：手动指定这些集群规范可能很乏味，特别是对于大型集群。咱们正在开发以编程方式启动任务的工具，例如使用像Kubernetes这样的集群管理器 。若是您想要查看支持的特定集群管理器，请提出一个 GitHub问题。

 

在模型中指定分布式设备

要对特定进程进行操做，您可使用与tf.device 用于指定运行在CPU或GPU上的操做相同的 功能。例如：

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:7"):   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://worker7.example.com:2222") as sess:   for _ in range(10000):     sess.run(train_op)

在上述示例中，在做业中的两个任务上建立变量，而且在做业中建立模型的计算密集型部分 。TensorFlow将插入做业之间的适当的数据传输（从到用于直传，以及从到用于施加梯度）。psworkerpsworkerworkerps

 

复印训练

一种称为“数据并行性”的常见培训配置涉及到worker在不一样小批量数据上对同一模型进行职业培训的多个任务，更新托管在ps 做业中的一个或多个任务中的共享参数。全部任务一般在不一样的机器上运行。有不少方法能够在TensorFlow中指定此结构，而且咱们正在构建库，以简化指定复制模型的工做。可能的方法包括：

• 图形内复制。在这种方法中，客户端构建一个 tf.Graph包含一组参数（在tf.Variable固定到节点中/job:ps）的参数; 和模型的计算密集型部分的多个副本，每一个副本都固定在不一样的任务中/job:worker。

• 图形间复制。在这种方法中，每一个/job:worker任务都有一个单独的客户端，一般与worker任务相同。每一个客户端构建一个包含参数的相似图（固定为 /job:ps使用前 tf.train.replica_device_setter 将其肯定性地映射到相同的任务）; 和模型的计算密集型部分的单个副本，固定到本地任务 /job:worker。

• 异步训练 在这种方法中，图的每一个副本都有一个独立的训练循环，无需协调地执行。它与上述两种复制形式兼容。

• 同步训练 在这种方法中，全部的副本都会读取当前参数的相同值，并行计算梯度，而后将它们应用在一块儿。它与图形内复制（例如使用CIFAR-10多GPU培训师中的梯度平均 ）和图形间复制（例如使用tf.train.SyncReplicasOptimizer）兼容 。

把它们放在一块儿：示例教练程序

如下代码显示了分布式教练程序的框架，实现了图形间复制和异步训练。它包括参数服务器和工做任务的代码。

import argparse import sys   import tensorflow as tf   FLAGS = None   def main(_):   ps_hosts = FLAGS.ps_hosts.split(",")   worker_hosts = FLAGS.worker_hosts.split(",")     # Create a cluster from the parameter server and worker hosts.   cluster = tf.train.ClusterSpec({"ps": ps_hosts, "worker": worker_hosts})     # Create and start a server for the local task.   server = tf.train.Server(cluster,                            job_name=FLAGS.job_name,                            task_index=FLAGS.task_index)     if FLAGS.job_name == "ps":     server.join()   elif FLAGS.job_name == "worker":       # Assigns ops to the local worker by default.     with tf.device(tf.train.replica_device_setter(         worker_device="/job:worker/task:%d" % FLAGS.task_index,         cluster=cluster)):         # Build model...       loss = ...       global_step = tf.contrib.framework.get_or_create_global_step()         train_op = tf.train.AdagradOptimizer(0.01).minimize(           loss, global_step=global_step)       # The StopAtStepHook handles stopping after running given steps.     hooks=[tf.train.StopAtStepHook(last_step=1000000)]       # The MonitoredTrainingSession takes care of session initialization,     # restoring from a checkpoint, saving to a checkpoint, and closing when done     # or an error occurs.     with tf.train.MonitoredTrainingSession(master=server.target,                                            is_chief=(FLAGS.task_index == 0),                                            checkpoint_dir="/tmp/train_logs",                                            hooks=hooks) as mon_sess:       while not mon_sess.should_stop():         # Run a training step asynchronously.         # See `tf.train.SyncReplicasOptimizer` for additional details on how to         # perform *synchronous* training.         # mon_sess.run handles AbortedError in case of preempted PS.         mon_sess.run(train_op)   if __name__ == "__main__":   parser = argparse.ArgumentParser()   parser.register("type", "bool", lambda v: v.lower() == "true")   # Flags for defining the tf.train.ClusterSpec   parser.add_argument(       "--ps_hosts",       type=str,       default="",       help="Comma-separated list of hostname:port pairs"   )   parser.add_argument(       "--worker_hosts",       type=str,       default="",       help="Comma-separated list of hostname:port pairs"   )   parser.add_argument(       "--job_name",       type=str,       default="",       help="One of 'ps', 'worker'"   )   # Flags for defining the tf.train.Server   parser.add_argument(       "--task_index",       type=int,       default=0,       help="Index of task within the job"   )   FLAGS, unparsed = parser.parse_known_args()   tf.app.run(main=main, argv=[sys.argv[0]] + unparsed)

要启动具备两个参数服务器和两个工做人员的培训师，请使用如下命令行（假设调用脚本）：trainer.py

# On ps0.example.com: $ python trainer.py \      --ps_hosts=ps0.example.com:2222,ps1.example.com:2222 \      --worker_hosts=worker0.example.com:2222,worker1.example.com:2222 \      --job_name=ps --task_index=0 # On ps1.example.com: $ python trainer.py \      --ps_hosts=ps0.example.com:2222,ps1.example.com:2222 \      --worker_hosts=worker0.example.com:2222,worker1.example.com:2222 \      --job_name=ps --task_index=1 # On worker0.example.com: $ python trainer.py \      --ps_hosts=ps0.example.com:2222,ps1.example.com:2222 \      --worker_hosts=worker0.example.com:2222,worker1.example.com:2222 \      --job_name=worker --task_index=0 # On worker1.example.com: $ python trainer.py \      --ps_hosts=ps0.example.com:2222,ps1.example.com:2222 \      --worker_hosts=worker0.example.com:2222,worker1.example.com:2222 \      --job_name=worker --task_index=1

词汇表

Client

客户端一般是一个构建TensorFlow图并构建一个 tensorflow::Session与集群进行交互的程序。客户端一般用Python或C ++编写。单个客户端进程能够直接与多个TensorFlow服务器进行交互（请参见上面的“复制培训”），单个服务器能够为多个客户端提供服务。

Cluster

TensorFlow集群包括一个或多个“做业”，每一个“做业”分为一个或多个“任务”的列表。集群一般专用于特定的高级目标，例如训练神经网络，并行使用许多机器。集群由tf.train.ClusterSpec对象定义。

Job

一份工做包括一份一般用于共同目的的“任务”清单。例如，名为ps（对于“参数服务器”）的做业一般会托管存储和更新变量的节点; 而名为“ workerjob” 的做业一般会承载执行计算密集型任务的无状态节点。做业中的任务一般在不一样的机器上运行。一组工做角色是灵活的：例如，a worker可能会保持一些状态。

Master service

提供远程访问一组分布式设备并充当会话目标的RPC服务。主服务实现 tensorflow::Session接口，负责协调一个或多个“工做服务”的工做。全部TensorFlow服务器都实现主服务。

Task

任务对应于特定的TensorFlow服务器，一般对应于单个进程。任务属于特定的“做业”，并由该做业的任务列表中的索引识别。

TensorFlow服务器运行tf.train.Server实例的进程，该实例是集群的成员，并导出“主服务”和“工做服务”。

Worker service

使用本地设备执行TensorFlow图形的一部分的RPC服务。worker服务实现了worker_service.proto。全部TensorFlow服务器都实现了工做服务。