今天將介紹如何創(chuàng)建 TensorFlow Server 集群，及如何在該集群中發(fā)布計(jì)算圖。在這里，我們假設(shè)您熟悉編寫低級(jí) TensorFlow 程序的基本概念 basic concepts。

你好，分布式 TensorFlow !

想要查看正在運(yùn)行的簡(jiǎn)單 TensorFlow 集群，請(qǐng)執(zhí)行以下操作：

# 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方法使用進(jìn)程內(nèi) Server 創(chuàng)建單進(jìn)程集群。

創(chuàng)建一個(gè)集群

TensorFlow “集群” 是一組參與 TensorFlow 圖的分布式執(zhí)行的 “任務(wù)”。每個(gè)任務(wù)都與 TensorFlow“Server” 相關(guān)聯(lián)，該 server 包含可用于創(chuàng)建會(huì)話的 “master 主干”，以及在圖中執(zhí)行操作的 “worker”。群集還可以劃分為一個(gè)或多個(gè) “作業(yè)”，其中每個(gè)作業(yè)包含一個(gè)或多個(gè)任務(wù)。

要?jiǎng)?chuàng)建群集，請(qǐng)?jiān)谌杭械拿總€(gè)任務(wù)中啟動(dòng)一個(gè) TensorFlow Server。每個(gè)任務(wù)通常在不同的計(jì)算機(jī)上運(yùn)行，但您可以在同一臺(tái)計(jì)算機(jī)上運(yùn)行多個(gè)任務(wù)（例如，控制不同的 GPU 設(shè)備）。在每項(xiàng)任務(wù)中，執(zhí)行以下操作：

創(chuàng)建一個(gè)描述集群中所有任務(wù)的 tf.train.ClusterSpec 每項(xiàng)任務(wù)都應(yīng)該相同

創(chuàng)建一個(gè) tf.train.Server，將 tf.train.ClusterSpec 傳遞給構(gòu)造函數(shù)，并使用作業(yè)名稱和任務(wù)索引標(biāo)識(shí)本地任務(wù)

創(chuàng)建一個(gè) tf.train.ClusterSpec 來(lái)描述集群

集群規(guī)范字典將作業(yè)名稱映射到網(wǎng)絡(luò)地址列表。將此字典傳遞給tf.train.ClusterSpec 構(gòu)造函數(shù)。例如：

在每個(gè)任務(wù)中創(chuàng)建一個(gè)tf.train.Server 實(shí)例

tf.train.Server 對(duì)象包含一組本地設(shè)備，一組與 tf.train.ClusterSpec 中其他任務(wù)的連接，以及一個(gè)可以使用它們執(zhí)行分布式計(jì)算的tf.Session。每個(gè) server 都是特定命名作業(yè)的成員，并且在該作業(yè)中具有任務(wù)索引。一個(gè) server 可以與群集中的任何其他服務(wù)器通信。

例如，要啟動(dòng)在 localhost：2222 和 localhost：2223 上運(yùn)行的兩個(gè)服務(wù)器的集群，請(qǐng)?jiān)诒镜赜?jì)算機(jī)上的兩個(gè)不同進(jìn)程中運(yùn)行以下代碼段：

# 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)

注意：手動(dòng)指定這些群集規(guī)范可能會(huì)很繁瑣，尤其是對(duì)于大型群集。我們正在開發(fā)以編程方式啟動(dòng)任務(wù)的工具，例如：使用像 Kubernetes 這樣的集群管理器。如果您希望獲得某些特定集群管理器的技術(shù)支持，請(qǐng)?jiān)?GitHub issue 提出。

在模型中指定分布式設(shè)備

要對(duì)特定進(jìn)程執(zhí)行操作，可以使用相同的 tf.device函數(shù)來(lái)指定操作是在 CPU 還是 GPU 上運(yùn)行。例如：

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 作業(yè)中的兩個(gè)任務(wù)上創(chuàng)建的，而模型的計(jì)算密集型部分是在 worker 作業(yè)中創(chuàng)建的。 TensorFlow 將在作業(yè)之間插入適當(dāng)?shù)臄?shù)據(jù)傳輸（從 ps 到 worker 用于正向傳遞，從 worker 到 ps 用于應(yīng)用漸變）。

復(fù)制訓(xùn)練

一個(gè)常見的訓(xùn)練配置，被稱為 “數(shù)據(jù)并行”，它涉及在不同的小批量數(shù)據(jù)上訓(xùn)練相同的模型額 worker 作業(yè)中的多個(gè)任務(wù)，更新在 ps 作業(yè)中的一個(gè)或多個(gè)任務(wù)中托管的共享參數(shù)。所有任務(wù)通常在不同的機(jī)器上運(yùn)行。在 TensorFlow 中有很多方法可以指定這個(gè)結(jié)構(gòu)，我們正在構(gòu)建庫(kù)，這將簡(jiǎn)化指定復(fù)制模型的工作?？赡苡行У姆椒òǎ?/p>

圖形內(nèi)復(fù)制。在這種方法中，客戶端構(gòu)建一個(gè)包含一組參數(shù)的tf.Graph（在tf.Variable 節(jié)點(diǎn)固定到 / job：ps）；以及模型中計(jì)算密集型部分的多個(gè)副本，每個(gè)副本都固定在 / job：worker 中的不同任務(wù)中

圖之間的復(fù)制。在這種方法中，每個(gè) / job：worker 任務(wù)都有一個(gè)單獨(dú)的客戶端，通常與 worker 任務(wù)在同一個(gè)進(jìn)程中。每個(gè)客戶端構(gòu)建一個(gè)包含參數(shù)的類似圖形（固定到 / job：psas，然后使用 tf.train.replica_device_setter將它們確定性地映射到相同的任務(wù)）；以及模型的計(jì)算密集型部分的單個(gè)副本，固定到 / job：worker 中的本地任務(wù)。

異步訓(xùn)練。在這種方法中，圖的每個(gè)副本都有一個(gè)獨(dú)立的訓(xùn)練循環(huán)，無(wú)需協(xié)調(diào)即可執(zhí)行。它與上述兩種復(fù)制形式兼容。

同步培訓(xùn)。在此方法中，所有副本都為當(dāng)前參數(shù)讀取相同的值，并行計(jì)算梯度，然后一起應(yīng)用。它與圖形內(nèi)復(fù)制兼容（例如，如在 CIFAR-10 multi-GPU trainer 中使用梯度平均），以及圖之間復(fù)制（例如，使用 tf.train.SyncReplicasOptimizer）均兼容。

綜合起來(lái)： 示例 trainer 計(jì)劃

以下代碼顯示了分布式 trainer 程序的框架，實(shí)現(xiàn)了圖形間復(fù)制和異步訓(xùn)練。它包括參數(shù) server 和 worker 任務(wù)的代碼。

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)

要使用兩個(gè)參數(shù) server 和兩個(gè) worker 程序啟動(dòng) trainer，請(qǐng)使用以下命令行（假設(shè)該腳本名為 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

詞匯表

客戶端

客戶端通常是一個(gè)程序，用于建立一個(gè) TensorFlow 圖，并構(gòu)造 tensorflow :: Session 以與集群交互?？蛻舳酥饕褂?Python 或 C ++ 編寫。一個(gè)單客戶端進(jìn)程可以直接與多個(gè) TensorFlow Server 交互（請(qǐng)參閱上面的 “復(fù)制訓(xùn)練”），一個(gè)單 Server 也可以為多個(gè)客戶端提供服務(wù)。

集群

一個(gè) TensorFlow 集群包括一個(gè)或多個(gè) “作業(yè)”，每個(gè) “作業(yè)” 被劃分為一個(gè)或多個(gè) “任務(wù)” 列表。一個(gè)集群通常效力于一個(gè)特定的高級(jí)別的目標(biāo)，例如多機(jī)并行訓(xùn)練一個(gè)神經(jīng)網(wǎng)絡(luò)，并行使用多臺(tái)機(jī)器。集群由 tf.train.ClusterSpec 目標(biāo)函數(shù)定義。

作業(yè)

一個(gè)作業(yè)由一系列 “任務(wù)” 組成，通常用于實(shí)現(xiàn)共同目的。 例如，名為 ps 的作業(yè)（用于 “參數(shù)服務(wù)器”）通常承載存儲(chǔ)和更新變量參數(shù)的節(jié)點(diǎn)；而名為 worker 的作業(yè)通常托管執(zhí)行計(jì)算密集型任務(wù)的無(wú)狀態(tài)節(jié)點(diǎn)。作業(yè)中的任務(wù)通常在不同的計(jì)算機(jī)上運(yùn)行。作業(yè)角色集是靈活的：例如，一個(gè) worker 可以維持某種狀態(tài)。

核心服務(wù)

RPC 服務(wù)為分布式設(shè)備之間提供遠(yuǎn)程訪問(wèn)，并充當(dāng)會(huì)話目標(biāo)的角色。核心設(shè)備實(shí)現(xiàn) tensorflow :: Session 接口，負(fù)責(zé)協(xié)調(diào)跨一個(gè)或多個(gè) “worker services” 之間的工作。所有 TensorFlow Servers 都實(shí)現(xiàn)核心服務(wù)。

任務(wù)

任務(wù)對(duì)應(yīng)于特定的 TensorFlow Server，通常對(duì)應(yīng)于單個(gè)進(jìn)程。任務(wù)屬于特定的 “作業(yè)”，并通過(guò)該作業(yè)的任務(wù)列表中的索引標(biāo)識(shí)來(lái)區(qū)分。

TensorFlowserver運(yùn)行 tf.train.Server 實(shí)例的進(jìn)程，該實(shí)例是集群的成員，并導(dǎo)出 “master service 主服務(wù)” 和 “worker service 工作服務(wù)”。

工作服務(wù)

RPC 服務(wù)，使用其本地設(shè)備執(zhí)行 TensorFlow 圖的部分。 Worker service 工作服務(wù)執(zhí)行 worker_service.proto。所有的 TensorFlow Server 都執(zhí)行 worker service 工作服務(wù)。