如何创建TensorFlow Server集群，及如何在该集群中发布计算图

今天将介绍如何创建 TensorFlow Server 集群，及如何在该集群中发布计算图。在这里，我们假设您熟悉编写低级 TensorFlow 程序的基本概念 basic concepts。

你好，分布式 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方法使用进程内 Server 创建单进程集群。

创建一个集群

TensorFlow “集群” 是一组参与 TensorFlow 图的分布式执行的 “任务”。每个任务都与 TensorFlow“Server” 相关联，该 server 包含可用于创建会话的 “master 主干”，以及在图中执行操作的 “worker”。群集还可以划分为一个或多个 “作业”，其中每个作业包含一个或多个任务。

要创建群集，请在群集中的每个任务中启动一个 TensorFlow Server。每个任务通常在不同的计算机上运行，但您可以在同一台计算机上运行多个任务（例如，控制不同的 GPU 设备）。在每项任务中，执行以下操作：

创建一个描述集群中所有任务的 tf.train.ClusterSpec 每项任务都应该相同

创建一个 tf.train.Server，将 tf.train.ClusterSpec 传递给构造函数，并使用作业名称和任务索引标识本地任务

创建一个 tf.train.ClusterSpec 来描述集群

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

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

tf.train.Server 对象包含一组本地设备，一组与 tf.train.ClusterSpec 中其他任务的连接，以及一个可以使用它们执行分布式计算的tf.Session。每个 server 都是特定命名作业的成员，并且在该作业中具有任务索引。一个 server 可以与群集中的任何其他服务器通信。

例如，要启动在 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 issue 提出。

在模型中指定分布式设备

要对特定进程执行操作，可以使用相同的 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)

在上面的示例中，变量是在 ps 作业中的两个任务上创建的，而模型的计算密集型部分是在 worker 作业中创建的。 TensorFlow 将在作业之间插入适当的数据传输（从 ps 到 worker 用于正向传递，从 worker 到 ps 用于应用渐变）。

复制训练

一个常见的训练配置，被称为 “数据并行”，它涉及在不同的小批量数据上训练相同的模型额 worker 作业中的多个任务，更新在 ps 作业中的一个或多个任务中托管的共享参数。所有任务通常在不同的机器上运行。在 TensorFlow 中有很多方法可以指定这个结构，我们正在构建库，这将简化指定复制模型的工作。可能有效的方法包括：

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

图之间的复制。在这种方法中，每个 / job：worker 任务都有一个单独的客户端，通常与 worker 任务在同一个进程中。每个客户端构建一个包含参数的类似图形（固定到 / job：psas，然后使用 tf.train.replica_device_setter将它们确定性地映射到相同的任务）；以及模型的计算密集型部分的单个副本，固定到 / job：worker 中的本地任务。

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

同步培训。在此方法中，所有副本都为当前参数读取相同的值，并行计算梯度，然后一起应用。它与图形内复制兼容（例如，如在 CIFAR-10 multi-GPU trainer 中使用梯度平均），以及图之间复制（例如，使用 tf.train.SyncReplicasOptimizer）均兼容。

综合起来： 示例 trainer 计划

以下代码显示了分布式 trainer 程序的框架，实现了图形间复制和异步训练。它包括参数 server 和 worker 任务的代码。

import argparseimport sysimport tensorflow as tfFLAGS = Nonedef 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)

要使用两个参数 server 和两个 worker 程序启动 trainer，请使用以下命令行（假设该脚本名为 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

词汇表

客户端

客户端通常是一个程序，用于建立一个 TensorFlow 图，并构造 tensorflow :: Session 以与集群交互。客户端主要使用 Python 或 C ++ 编写。一个单客户端进程可以直接与多个 TensorFlow Server 交互（请参阅上面的 “复制训练”），一个单 Server 也可以为多个客户端提供服务。

集群

一个 TensorFlow 集群包括一个或多个 “作业”，每个 “作业” 被划分为一个或多个 “任务” 列表。一个集群通常效力于一个特定的高级别的目标，例如多机并行训练一个神经网络，并行使用多台机器。集群由 tf.train.ClusterSpec 目标函数定义。

作业

一个作业由一系列 “任务” 组成，通常用于实现共同目的。 例如，名为 ps 的作业（用于 “参数服务器”）通常承载存储和更新变量参数的节点；而名为 worker 的作业通常托管执行计算密集型任务的无状态节点。作业中的任务通常在不同的计算机上运行。作业角色集是灵活的：例如，一个 worker 可以维持某种状态。

核心服务

RPC 服务为分布式设备之间提供远程访问，并充当会话目标的角色。核心设备实现 tensorflow :: Session 接口，负责协调跨一个或多个 “worker services” 之间的工作。所有 TensorFlow Servers 都实现核心服务。

任务

任务对应于特定的 TensorFlow Server，通常对应于单个进程。任务属于特定的 “作业”，并通过该作业的任务列表中的索引标识来区分。

TensorFlowserver运行 tf.train.Server 实例的进程，该实例是集群的成员，并导出 “master service 主服务” 和 “worker service 工作服务”。

工作服务

RPC 服务，使用其本地设备执行 TensorFlow 图的部分。 Worker service 工作服务执行 worker_service.proto。所有的 TensorFlow Server 都执行 worker service 工作服务。