深入浅出TensorFlow（七）TensorFlow计算加速_语言 & 开发_郑泽宇

最新发布《数智时代的AI人才粮仓模型解读白皮书（2024版）》，立即领取! 了解详情 



 写点什么

2017 年 2 月 16 日，Google 正式对外发布 Google TensorFlow 1.0 版本，并保证本次的发布版本 API 接口完全满足生产环境稳定性要求。这是 TensorFlow 的一个重要里程碑，标志着它可以正式在生产环境放心使用。在国内，从 InfoQ 的判断来看，TensorFlow 仍处于创新传播曲线的创新者使用阶段，大部分人对于 TensorFlow 还缺乏了解，社区也缺少帮助落地和使用的中文资料。InfoQ 期望通过深入浅出 TensorFlow 系列文章能够推动 Tensorflow 在国内的发展。欢迎加入 QQ 群（群号：183248479）深入讨论和交流。下面为本系列的前六篇文章：

深入浅出Tensorflow（一）：深度学习及TensorFlow 简介

深入浅出TensorFlow（二）：TensorFlow 解决MNIST 问题入门

深入浅出Tensorflow（三）：训练神经网络模型的常用方法

深入浅出Tensorflow（四）：卷积神经网络

深入浅出Tensorflow（五）：循环神经网络简介

深入浅出TensorFlow（六）TensorFlow 高层封装

在前面的文章中介绍了使用TensorFlow 实现各种深度学习的算法。然而要将深度学习应用到实际问题中，一个非常大的问题在于训练深度学习模型需要的计算量太大。比如要将Inception-v3 模型在单机单卡上训练到78% 的正确率需要将近半年的时间，这样的训练速度是完全无法应用到实际生产中的。为了加速训练过程，本文将介绍如何通过TensorFlow 利用GPU 或/ 和分布式计算进行模型训练。

TensorFlow 使用 GPU

TensorFlow 程序可以通过 tf.device 函数来指定运行每一个操作的设备，这个设备可以是本地的 CPU 或者 GPU，也可以是某一台远程的服务器。TensorFlow 会给每一个可用的设备一个名称，tf.device 函数可以通过设备的名称来指定执行运算的设备。比如 CPU 在 TensorFlow 中的名称为 /cpu:0。

在默认情况下，即使机器有多个 CPU，TensorFlow 也不会区分它们，所有的 CPU 都使用 /cpu:0 作为名称。而一台机器上不同 GPU 的名称是不同的，第 n 个 GPU 在 TensorFlow 中的名称为 /gpu:n。比如第一个 GPU 的名称为 /gpu:0，第二个 GPU 名称为 /gpu:1，以此类推。

TensorFlow 提供了一个快捷的方式来查看运行每一个运算的设备。在生成会话时，可以通过设置 log_device_placement 参数来打印运行每一个运算的设备。下面的程序展示了如何使用 log_device_placement 这个参数。

复制代码

  
import tensorflow as tf
 
a = tf.constant([1.0, 2.0, 3.0], shape=[3], name='a')
b = tf.constant([1.0, 2.0, 3.0], shape=[3], name='b')
c = a + b
# 通过 log_device_placement 参数来输出运行每一个运算的设备。
sess = tf.Session(config=tf.ConfigProto(log_device_placement=True))
print sess.run(c)
 
'''
  在没有 GPU 的机器上运行以上代码可以得到以下输出：
Device mapping: no known devices.
 
add: /job:localhost/replica:0/task:0/cpu:0
b: /job:localhost/replica:0/task:0/cpu:0
a: /job:localhost/replica:0/task:0/cpu:0
[ 2.  4.  6.]
'''

在以上代码中，TensorFlow 程序生成会话时加入了参数 log_device_placement=True，所以程序会将运行每一个操作的设备输出到屏幕。于是除了可以看到最后的计算结果之外，还可以看到类似“add:/job:localhost/replica:0/task:0/cpu:0”这样的输出。这些输出显示了执行每一个运算的设备。比如加法操作 add 是通过 CPU 来运行的，因为它的设备名称中包含了 /cpu:0。

在配置好 GPU 环境的 TensorFlow 中，如果操作没有明确地指定运行设备，那么 TensorFlow 会优先选择 GPU。比如将以上代码在亚马逊（Amazon Web Services, AWS）的 g2.8xlarge 实例上运行时，会得到以下运行结果。

复制代码

  
Device mapping:
/job:localhost/replica:0/task:0/gpu:0 -> device: 0, name: GRID K520, pci bus id: 0000:00:03.0
/job:localhost/replica:0/task:0/gpu:1 -> device: 1, name: GRID K520, pci bus id: 0000:00:04.0
/job:localhost/replica:0/task:0/gpu:2 -> device: 2, name: GRID K520, pci bus id: 0000:00:05.0
/job:localhost/replica:0/task:0/gpu:3 -> device: 3, name: GRID K520, pci bus id: 0000:00:06.0
 
add: /job:localhost/replica:0/task:0/gpu:0
b: /job:localhost/replica:0/task:0/gpu:0
a: /job:localhost/replica:0/task:0/gpu:0
[ 2.  4.  6.]

从上面的输出可以看到在配置好 GPU 环境的 TensorFlow 中，TensorFlow 会自动优先将运算放置在 GPU 上。不过，尽管 g2.8xlarge 实例有 4 个 GPU，在默认情况下，TensorFlow 只会将运算优先放到 /gpu:0 上。于是可以看见在上面的程序中，所有的运算都被放在了 /gpu:0 上。如果需要将某些运算放到不同的 GPU 或者 CPU 上，就需要通过 tf.device 来手工指定。下面的程序给出了一个通过 tf.device 手工指定运行设备的样例。

复制代码

  
import tensorflow as tf
 
# 通过 tf.device 将运算指定到特定的设备上。
with tf.device('/cpu:0'):
   a = tf.constant([1.0, 2.0, 3.0], shape=[3], name='a')
   b = tf.constant([1.0, 2.0, 3.0], shape=[3], name='b')
with tf.device('/gpu:1'):
   c = a + b
 
sess = tf.Session(config=tf.ConfigProto(log_device_placement=True))
print sess.run(c)
 
'''
在 AWS g2.8xlarge 实例上运行上述代码可以得到一下结果：
Device mapping:
/job:localhost/replica:0/task:0/gpu:0 -> device: 0, name: GRID K520, pci bus id: 0000:00:03.0
/job:localhost/replica:0/task:0/gpu:1 -> device: 1, name: GRID K520, pci bus id: 0000:00:04.0
/job:localhost/replica:0/task:0/gpu:2 -> device: 2, name: GRID K520, pci bus id: 0000:00:05.0
/job:localhost/replica:0/task:0/gpu:3 -> device: 3, name: GRID K520, pci bus id: 0000:00:06.0
{1}
add: /job:localhost/replica:0/task:0/gpu:1
b: /job:localhost/replica:0/task:0/cpu:0
a: /job:localhost/replica:0/task:0/cpu:0
[ 2.  4.  6.]
'''

在以上代码中可以看到生成常量 a 和 b 的操作被加载到了 CPU 上，而加法操作被放到了第二个 GPU“/gpu:1”上。在 TensorFlow 中，不是所有的操作都可以被放在 GPU 上，如果强行将无法放在 GPU 上的操作指定到 GPU 上，那么程序将会报错。

多 GPU 并行

下面将给出具体的 TensorFlow 代码在一台机器的多个 GPU 上并行训练深度学习模型。因为一般来说一台机器上的多个 GPU 性能相似，所以在这种设置下会更多地采用同步模式训练深度学习模型。下面将给出具体的代码，在多 GPU 上训练深度学习模型解决 MNIST 问题。（该代码可以在 TensorFlow 0.9.0 下运行，对于更新 TensorFlow 的版本，请参考 Github 代码库： https://github.com/caicloud/tensorflow-tutorial ）

复制代码

 
# -*- coding: utf-8 -*-
 
from datetime import datetime
import os
import time
 
import tensorflow as tf
# 定义训练神经网络时需要用到的配置。
BATCH_SIZE = 100 
LEARNING_RATE_BASE = 0.001
LEARNING_RATE_DECAY = 0.99
REGULARAZTION_RATE = 0.0001
TRAINING_STEPS = 1000
MOVING_AVERAGE_DECAY = 0.99 
N_GPU = 4
# 定义日志和模型输出的路径。
MODEL_SAVE_PATH = "/path/to/logs_and_models/"
MODEL_NAME = "model.ckpt"
# 定义数据存储的路径。因为需要为不同的 GPU 提供不同的训练数据，所以通过 placerholder
# 的方式就需要手动准备多份数据。为了方便训练数据的获取过程，可以采用输入队列的方式从
# TFRecord 中读取数据。于是在这里提供的数据文件路径为将 MNIST 训练数据转化为
# TFRecords 格式之后的路径。
DATA_PATH = "/path/to/data.tfrecords" 
# 定义输入队列得到训练数据，具体细节可以参考《TensorFlow: 实战 Google 深度学习框架》
# 第七章。
def get_input():
   filename_queue = tf.train.string_input_producer([DATA_PATH]) 
   reader = tf.TFRecordReader()
_, serialized_example = reader.read(filename_queue)
# 定义数据解析格式。
   features = tf.parse_single_example(
       serialized_example,
       features={
           'image_raw': tf.FixedLenFeature([], tf.string),
           'pixels': tf.FixedLenFeature([], tf.int64),
           'label': tf.FixedLenFeature([], tf.int64),
       })        
   # 解析图片和标签信息。
   decoded_image = tf.decode_raw(features['image_raw'], tf.uint8)
   reshaped_image = tf.reshape(decoded_image, [784])
   retyped_image = tf.cast(reshaped_image, tf.float32)
   label = tf.cast(features['label'], tf.int32)
   # 定义输入队列并返回。
   min_after_dequeue = 10000
   capacity = min_after_dequeue + 3 * BATCH_SIZE
return tf.train.shuffle_batch(
       [retyped_image, label], 
       batch_size=BATCH_SIZE, 
       capacity=capacity, 
       min_after_dequeue=min_after_dequeue)
# 定义损失函数。对于给定的训练数据、正则化损失计算规则和命名空间，计算在这个命名空间
# 下的总损失。之所以需要给定命名空间是因为不同的 GPU 上计算得出的正则化损失都会加入名为
# loss 的集合，如果不通过命名空间就会将不同 GPU 上的正则化损失都加进来。
def get_loss(x, y_, regularizer, scope):
   # 沿用第四篇文章中定义的卷积神经网络计算前向传播结果。
y = inference(x, regularizer)
# 计算交叉熵损失。
cross_entropy = tf.reduce_mean(
   tf.nn.sparse_softmax_cross_entropy_with_logits(y, y_))
# 计算当前 GPU 上计算得到的正则化损失。
regularization_loss = tf.add_n(tf.get_collection('losses', scope))
# 计算最终的总损失。
   loss = cross_entropy + regularization_loss
   return loss
   # 计算每一个变量梯度的平均值。
def average_gradients(tower_grads):
average_grads = []
# 枚举所有的变量和变量在不同 GPU 上计算得出的梯度。
for grad_and_vars in zip(*tower_grads):
         # 计算所有 GPU 上的梯度平均值。
         grads = []
         for g, _ in grad_and_vars:
             expanded_g = tf.expand_dims(g, 0)
             grads.append(expanded_g)
         grad = tf.concat(0, grads)
         grad = tf.reduce_mean(grad, 0)
 
         v = grad_and_vars[0][7]
         grad_and_var = (grad, v)
         # 将变量和它的平均梯度对应起来。
         average_grads.append(grad_and_var)
   # 返回所有变量的平均梯度，这将被用于变量更新。
   return average_grads
# 主训练过程。
def main(argv=None): 
   # 将简单的运算放在 CPU 上，只有神经网络的训练过程放在 GPU 上。
with tf.Graph().as_default(), tf.device('/cpu:0'):
       # 获取训练 batch。
       x, y_ = get_input()
       regularizer = tf.contrib.layers.l2_regularizer(REGULARAZTION_RATE)
       # 定义训练轮数和指数衰减的学习率。
   global_step = tf.get_variable(
       'global_step', [], initializer=tf.constant_initializer(0), 
        trainable=False)
       learning_rate = tf.train.exponential_decay(
           LEARNING_RATE_BASE, global_step, 60000 / BATCH_SIZE, 
           LEARNING_ RATE_DECAY)       
       # 定义优化方法。
       opt = tf.train.GradientDescentOptimizer(learning_rate)
 
       tower_grads = []
       # 将神经网络的优化过程跑在不同的 GPU 上。
       for i in range(N_GPU):
           # 将优化过程指定在一个 GPU 上。
           with tf.device('/gpu:%d' % i):
               with tf.name_scope('GPU_%d' % i) as scope:
                   cur_loss = get_loss(x, y_, regularizer, scope)
                   # 在第一次声明变量之后，将控制变量重用的参数设置为 True。这样可以
                   # 让不同的 GPU 更新同一组参数。注意 tf.name_scope 函数并不会影响
   # tf.get_ variable 的命名空间。
                   tf.get_variable_scope().reuse_variables()
                   # 使用当前 GPU 计算所有变量的梯度。
                   grads = opt.compute_gradients(cur_loss)
                   tower_grads.append(grads)
       # 计算变量的平均梯度，并输出到 TensorBoard 日志中。
       grads = average_gradients(tower_grads)
       for grad, var in grads:
           if grad is not None:
               tf.histogram_summary(
                   'gradients_on_average/%s' % var.op.name, grad)
       # 使用平均梯度更新参数。
       apply_gradient_op = opt.apply_gradients(
           grads, global_step=global_ step)
       for var in tf.trainable_variables():
           tf.histogram_summary(var.op.name, var)
       # 计算变量的滑动平均值。
       variable_averages = tf.train.ExponentialMovingAverage(
           MOVING_AVERAGE_DECAY, global_step)
       variables_averages_op = variable_averages.apply(
           tf.trainable_variables())
       # 每一轮迭代需要更新变量的取值并更新变量的滑动平均值。
       train_op = tf.group(apply_gradient_op, variables_averages_op)
 
       saver = tf.train.Saver(tf.all_variables())
       summary_op = tf.merge_all_summaries()        
       init = tf.initialize_all_variables()
       # 训练过程。
       with tf.Session(config=tf.ConfigProto(
           allow_soft_placement=True, 
           log_device_placement=True)) as sess:
           # 初始化所有变量并启动队列。
           init.run()
           coord = tf.train.Coordinator()
           threads = tf.train.start_queue_runners(sess=sess, coord=coord)
           summary_writer = tf.train.SummaryWriter(
               MODEL_SAVE_PATH, sess.graph)
 
           for step in range(TRAINING_STEPS):
               # 执行神经网络训练操作，并记录训练操作的运行时间。
               start_time = time.time()
               _, loss_value = sess.run([train_op, cur_loss])
               duration = time.time() - start_time
               # 每隔一段时间展示当前的训练进度，并统计训练速度。
               if step != 0 and step % 10 == 0:
                   # 计算使用过的训练数据个数。因为在每一次运行训练操作时，每一个 GPU 
                   # 都会使用一个 batch 的训练数据，所以总共用到的训练数据个数为
                   # batch 大小×GPU 个数。
                   num_examples_per_step = BATCH_SIZE * N_GPU
                   # num_examples_per_step 为本次迭代使用到的训练数据个数， 
                   # duration 为运行当前训练过程使用的时间，于是平均每秒可以处理的训
                   # 练数据个数为 num_examples_per_step / duration。
                   examples_per_sec = num_examples_per_step / duration
                   # duration 为运行当前训练过程使用的时间，因为在每一个训练过程中， 
                   # 每一个 GPU 都会使用一个 batch 的训练数据，所以在单个 batch 上的训
                   # 练所需要时间为 duration / GPU 个数。
                   sec_per_batch = duration / N_GPU
                   # 输出训练信息。
                   format_str = ('step %d, loss = %.2f (%.1f examples/ '
                                   ' sec; %.3f sec/batch)')
                   print(format_str % (step, loss_value, 
                                           examples_per_sec, sec_per_batch))
                   # 通过 TensorBoard 可视化训练过程。
                   summary = sess.run(summary_op)
                   summary_writer.add_summary(summary, step)
               # 每隔一段时间保存当前的模型。
               if step % 1000 == 0 or (step + 1) == TRAINING_STEPS:
                   checkpoint_path = os.path.join(
                       MODEL_SAVE_PATH, MODEL_ NAME)
                   saver.save(sess, checkpoint_path, global_step=step)
 
           coord.request_stop()
           coord.join(threads)
 
if __name__ == '__main__':
   tf.app.run()
 
'''
在 AWS 的 g2.8xlarge 实例上运行上面这段程序可以得到类似下面的结果：
step 10, loss = 71.90 (15292.3 examples/sec; 0.007 sec/batch)
step 20, loss = 37.97 (18758.3 examples/sec; 0.005 sec/batch)
step 30, loss = 9.54 (16313.3 examples/sec; 0.006 sec/batch)
step 40, loss = 11.84 (14199.0 examples/sec; 0.007 sec/batch)
...
step 980, loss = 0.66 (15034.7 examples/sec; 0.007 sec/batch)
step 990, loss = 1.56 (16134.1 examples/sec; 0.006 sec/batch)
'''

(点击放大图像)

图1 在AWS 的g2.8xlarge 实例上运行MNIST 样例程序时GPU 的使用情况

在AWS 的g2.8xlarge 实例上运行以上代码可以同时使用4 个GPU 训练神经网络。图1 显示了运行样例代码时不同GPU 的使用情况。因为运行的神经网络规模比较小，所以在图1 中显示的GPU 使用率不高。如果训练大型的神经网络模型，TensorFlow 将会占满所有用到的GPU。

(点击放大图像)

图 2 使用了 4 个 GPU 的 TensorFlow 计算图可视化结果

图 2 展示了通过 TensorBoard 可视化得到的样例代码 TensorFlow 计算图，其中节点上的颜色代表了不同的设备，比如黑色代表 CPU、白色代表第一个 GPU，等等。从图 2 中可以看出，训练神经网络的主要过程被放到了 GPU_0、GPU_1、GPU_2 和 GPU_3 这 4 个模块中，而且每一个模块运行在一个 GPU 上。

分布式 TensorFlow

通过多 GPU 并行的方式可以达到很好的加速效果。然而一台机器上能够安装的 GPU 有限，要进一步提升深度学习模型的训练速度，就需要将 TensorFlow 分布式运行在多台机器上。以下代码展示了如何创建一个最简单的 TensorFlow 集群：

复制代码

  
import tensorflow as tf
c = tf.constant("Hello, distributed TensorFlow!")
# 创建一个本地 TensorFlow 集群
server = tf.train.Server.create_local_server()  
# 在集群上创建一个会话。
sess = tf.Session(server.target)      
# 输出 Hello, distributed TensorFlow!
print sess.run(c)

在以上代码中，首先通过 tf.train.Server.create_local_server 函数在本地建立了一个只有一台机器的 TensorFlow 集群。然后在该集群上生成了一个会话，并通过生成的会话将运算运行在创建的 TensorFlow 集群上。虽然这只是一个单机集群，但它大致反应了 TensorFlow 集群的工作流程。TensorFlow 集群通过一系列的任务（tasks）来执行 TensorFlow 计算图中的运算。一般来说，不同任务跑在不同机器上。最主要的例外是使用 GPU 时，不同任务可以使用同一台机器上的不同 GPU。TensorFlow 集群中的任务也会被聚合成工作（jobs），每个工作可以包含一个或者多个任务。比如在训练深度学习模型时，一台运行反向传播的机器是一个任务，而所有运行反向传播机器的集合是一种工作。

上面的样例代码是只有一个任务的集群。当一个 TensorFlow 集群有多个任务时，需要使用 tf.train.ClusterSpec 来指定运行每一个任务的机器。比如以下代码展示了在本地运行有两个任务的 TensorFlow 集群。第一个任务的代码如下：

复制代码

  
import tensorflow as tf
c = tf.constant("Hello from server1!")
 
# 生成一个有两个任务的集群，一个任务跑在本地 2222 端口，另外一个跑在本地 2223 端口。
cluster = tf.train.ClusterSpec(
   {"local": ["localhost:2222", "localhost: 2223"]})
# 通过上面生成的集群配置生成 Server，并通过 job_name 和 task_index 指定当前所启动
# 的任务。因为该任务是第一个任务，所以 task_index 为 0。
server = tf.train.Server(cluster, job_name="local", task_index=0)
 
# 通过 server.target 生成会话来使用 TensorFlow 集群中的资源。通过设置
# log_device_placement 可以看到执行每一个操作的任务。
sess = tf.Session(
   server.target, config=tf.ConfigProto(log_device_placement=True)) 
print sess.run(c)
server.join()

下面给出了第二个任务的代码：

复制代码

  
import tensorflow as tf
c = tf.constant("Hello from server2!")
 
# 和第一个程序一样的集群配置。集群中的每一个任务需要采用相同的配置。
cluster = tf.train.ClusterSpec(
   {"local": ["localhost:2222", "localhost: 2223"]})
# 指定 task_index 为 1，所以这个程序将在 localhost:2223 启动服务。
server = tf.train.Server(cluster, job_name="local", task_index=1)
# 剩下的代码都和第一个任务的代码一致。
...

启动第一个任务后，可以得到类似下面的输出：

复制代码

  
I tensorflow/core/distributed_runtime/rpc/grpc_channel.cc:206] Initialize HostPortsGrpcChannelCache for job local -> {localhost:2222, localhost:2223}
I tensorflow/core/distributed_runtime/rpc/grpc_server_lib.cc:202] Started server with target: grpc://localhost:2222
E1123 08:26:06.824503525   12232 tcp_client_posix.c:173]     failed to connect to 'ipv4:127.0.0.1:2223': socket error: connection refused
E1123 08:26:08.825022513   12232 tcp_client_posix.c:173]     failed to connect to 'ipv4:127.0.0.1:2223': socket error: connection refused
I tensorflow/core/common_runtime/simple_placer.cc:818] Const: /job:local/ replica:0/task:0/cpu:0
Const: /job:local/replica:0/task:0/cpu:0
Hello from server1!

从第一个任务的输出中可以看到，当只启动第一个任务时，程序会停下来等待第二个任务启动，而且持续输出 failed to connect to ‘ipv4:127.0.0.1:2223’: socket error: connection refused。当第二个任务启动后，可以看到从第一个任务中会输出 Hello from server1! 的结果。第二个任务的输出如下：

复制代码

  
I tensorflow/core/distributed_runtime/rpc/grpc_channel.cc:206] Initialize HostPortsGrpcChannelCache for job local -> {localhost:2222, localhost:2223} 
I tensorflow/core/distributed_runtime/rpc/grpc_server_lib.cc:202] Started server with target: grpc://localhost:2223 
Const: /job:local/replica:0/task:0/cpu:0 
I tensorflow/core/common_runtime/simple_placer.cc:818] Const: /job:local/ replica:0/task:0/cpu:0 
Hello from server2!

值得注意的是第二个任务中定义的计算也被放在了设备 /job:local/replica:0/task:0/cpu:0 上。也就是说这个计算将由第一个任务来执行。从上面这个样例可以看到，通过 tf.train.Server.target 生成的会话可以统一管理整个 TensorFlow 集群中的资源。

和使用多 GPU 类似，TensorFlow 支持通过 tf.device 来指定操作运行在哪个任务上。比如将第二个任务中定义计算的语句改为以下代码，就可以看到这个计算将被调度到 /job:local/replica:0/task:1/cpu:0 上面。

复制代码

  
with tf.device("/job:local/task:1"):
   c = tf.constant("Hello from server2!")

在上面的样例中只定义了一个工作“local”。但一般在训练深度学习模型时，会定义两个工作。一个工作专门负责存储、获取以及更新变量的取值，这个工作所包含的任务统称为参数服务器（parameter server，ps）。另外一个工作负责运行反向传播算法来获取参数梯度，这个工作所包含的任务统称为计算服务器（worker）。下面给出了一个比较常见的用于训练深度学习模型的 TensorFlow 集群配置方法。

复制代码

  
tf.train.ClusterSpec({
   "worker": [
       "tf-worker0:2222",
       "tf-worker1:2222",
       "tf-worker2:2222"
   ],
   "ps": [
       "tf-ps0:2222",
       "tf-ps1:2222"
   ]})

使用分布式 TensorFlow 训练深度学习模型一般有两种方式。一种方式叫做计算图内分布式（in-graph replication）。使用这种分布式训练方式时，所有的任务都会使用一个 TensorFlow 计算图中的变量（也就是深度学习模型中的参数），而只是将计算部分发布到不同的计算服务器上。

上面给出的使用多 GPU 样例程序就是这种方式。多 GPU 样例程序将计算复制了多份，每一份放到一个 GPU 上进行运算。但不同的 GPU 使用的参数都是在一个 TensorFlow 计算图中的。因为参数都是存在同一个计算图中，所以同步更新参数比较容易控制。在上面给出的代码也实现了参数的同步更新。然而因为计算图内分布式需要有一个中心节点来生成这个计算图并分配计算任务，所以当数据量太大时，这个中心节点容易造成性能瓶颈。

另外一种分布式 TensorFlow 训练深度学习模型的方式叫计算图之间分布式（between-graph replication）。使用这种分布式方式时，在每一个计算服务器上都会创建一个独立的 TensorFlow 计算图，但不同计算图中的相同参数需要以一种固定的方式放到同一个参数服务器上。TensorFlow 提供了 tf.train.replica_device_setter 函数来帮助完成这一个过程。

因为每个计算服务器的 TensorFlow 计算图是独立的，所以这种方式的并行度要更高。但在计算图之间分布式下进行参数的同步更新比较困难。

为了解决这个问题，TensorFlow 提供了 tf.train.SyncReplicasOptimizer 函数来帮助实现参数的同步更新。这让计算图之间分布式方式被更加广泛地使用。因为篇幅所限，所以这里不在给出具体代码，具体代码可以在 Github 代码库中找到。

使用 Caicloud TaaS 平台运行分布式 TensorFlow

每次运行分布式 TensorFlow 都需要登录不同的机器来启动集群。这使得使用起来非常不方便。当需要使用 100 台机器运行分布式 TensorFlow 时，需要手动登录到每一台机器并启动 TensorFlow 服务，这个过程十分繁琐。而且，当某个服务器上的程序死掉之后，TensorFlow 并不能自动重启，这给监控工作带来了巨大的难度。

如果类比 TensorFlow 与 Hadoop，可以发现 TensorFlow 只实现了相当于 Hadoop 中 MapReduce 的计算框架，而没有提供类似 Yarn 的集群管理工具以及 HDFS 的存储系统。为了降低分布式 TensorFlow 的使用门槛，才云科技 (Caicloud.io) 基于 Kubernetes 容器云平台提供了一个分布式 TensorFlow 平台 TensorFlow as a Service（TaaS）。

从我们提供的开源代码库中可以看出，编写分布式 TensorFlow 程序需要指定很多与模型训练无关的代码来完成 TensorFlow 集群的设置工作。为了降低分布式 TensorFlow 的学习成本，Caicloud 的 TensorFlow as a Service（TaaS）平台首先对 TensorFlow 集群进行了更高层的封装，屏蔽了其中与模型训练无关的底层细节。

其次，TaaS 平台结合了谷歌开源的容器云平台管理工具 Kubernetes 来实现对分布式 TensorFlow 任务的管理和监控，并支持通过 UI 设置分布式 TensorFlow 任务的节点个数、是否使用 GPU 等信息。

有关 Caicloud TaaS 平台的更多信息可以了解。

有关 TaaS 平台的使用文档可以参考。

本文内容来自作者图书作品《TensorFlow: 实战 Google 深度学习框架》，点击购买。

作者介绍

郑泽宇，才云首席大数据科学家，前谷歌高级工程师。从 2013 年加入谷歌至今，郑泽宇作为主要技术人员参与并领导了多个大数据项目，拥有丰富机器学习、数据挖掘工业界及科研项目经验。2014 年，他提出产品聚类项目用于衔接谷歌购物和谷歌知识图谱（Knowledge Graph）数据，使得知识卡片形式的广告逐步取代传统的产品列表广告，开启了谷歌购物广告在搜索页面投递的新纪元。他于 2013 年 5 月获得美国 Carnegie Mellon University（CMU）大学计算机硕士学位, 期间在顶级国际学术会议上发表数篇学术论文，并获得西贝尔奖学金。

公众号推荐：

跳进 AI 的奇妙世界，一起探索未来工作的新风貌！想要深入了解 AI 如何成为产业创新的新引擎？好奇哪些城市正成为 AI 人才的新磁场？《中国生成式 AI 开发者洞察 2024》由 InfoQ 研究中心精心打造，为你深度解锁生成式 AI 领域的最新开发者动态。无论你是资深研发者，还是对生成式 AI 充满好奇的新手，这份报告都是你不可错过的知识宝典。欢迎大家扫码关注「AI前线」公众号，回复「开发者洞察」领取。

发布

暂无评论

创作场景

深入浅出 TensorFlow（七）TensorFlow 计算加速

TensorFlow 使用 GPU

多 GPU 并行

分布式 TensorFlow

使用 Caicloud TaaS 平台运行分布式 TensorFlow

作者介绍

公众号推荐：

评论

再谈前端性能监控及4个最佳工具分享

开发中Docker常用容器记录

到底哪种类型的错误信息会阻止business transaction的保存

一个查看 SAP UI5 控件所有公有方法的小技巧

如何在SAP CRM WebClient UI里创建HANA Live Report

Inner Join, Left Outer Join和Association的区别

Shell脚本-简单爬虫

☕【Java技术之旅】从底层角度去认识线程的原理

C4C Cloud Application Studio做ABSL开发的一些性能方面的最佳实践

如何将BSP应用配置成Fiori Launchpad上的一个tile

SAP 不同 ABAP 系统里同一 Customizing activity 的显示差异分析

另一种方式实现事务码SE16里的结果集修改

架构实战营-作业四

Go 并发编程 — 深入浅出 sync.Pool ，围观最全的使用姿势，理解最深刻的原理

基于Mac的手动搭建WordPress个人站点的方法

如何根据日志来了解一个请求经历了什么？

架构实战营模块4作业

找出 SAP OData service出错根源的小技巧

模块四-千万级学生管理系统试卷存储方案

SAP Fiori 应用 Footerbar 区域按钮的高亮显示逻辑

利用Chrome的Heap Snapshot功能分析一个时间段内的内存占用率

自我复盘

高性能 JavaScriptの五 -- 快响应用户界面

MySQL主从网络延迟解决方案

🚄【Redis 干货领域】帮你完全搞定 Cluster 原理（实践篇）

Python 持久化 - 文件

Go 并发编程-goroutine 初体验

Authorization object在哪些ABAP代码里使用到

设计千万级学生管理系统的考试试卷存储方案

和另一半过不一样的520

SAP Fiori Launchpad Tile点击后跳转的调试技巧


	import tensorflow as tf

	a = tf.constant([1.0, 2.0, 3.0], shape=[3], name='a')
	b = tf.constant([1.0, 2.0, 3.0], shape=[3], name='b')
	c = a + b
	# 通过 log_device_placement 参数来输出运行每一个运算的设备。
	sess = tf.Session(config=tf.ConfigProto(log_device_placement=True))
	print sess.run(c)

	'''
	在没有 GPU 的机器上运行以上代码可以得到以下输出：
	Device mapping: no known devices.

	add: /job:localhost/replica:0/task:0/cpu:0
	b: /job:localhost/replica:0/task:0/cpu:0
	a: /job:localhost/replica:0/task:0/cpu:0
	[ 2. 4. 6.]
	'''


	Device mapping:
	/job:localhost/replica:0/task:0/gpu:0 -> device: 0, name: GRID K520, pci bus id: 0000:00:03.0
	/job:localhost/replica:0/task:0/gpu:1 -> device: 1, name: GRID K520, pci bus id: 0000:00:04.0
	/job:localhost/replica:0/task:0/gpu:2 -> device: 2, name: GRID K520, pci bus id: 0000:00:05.0
	/job:localhost/replica:0/task:0/gpu:3 -> device: 3, name: GRID K520, pci bus id: 0000:00:06.0

	add: /job:localhost/replica:0/task:0/gpu:0
	b: /job:localhost/replica:0/task:0/gpu:0
	a: /job:localhost/replica:0/task:0/gpu:0
	[ 2. 4. 6.]


	import tensorflow as tf

	# 通过 tf.device 将运算指定到特定的设备上。
	with tf.device('/cpu:0'):
	a = tf.constant([1.0, 2.0, 3.0], shape=[3], name='a')
	b = tf.constant([1.0, 2.0, 3.0], shape=[3], name='b')
	with tf.device('/gpu:1'):
	c = a + b

	sess = tf.Session(config=tf.ConfigProto(log_device_placement=True))
	print sess.run(c)

	'''
	在 AWS g2.8xlarge 实例上运行上述代码可以得到一下结果：
	Device mapping:
	/job:localhost/replica:0/task:0/gpu:0 -> device: 0, name: GRID K520, pci bus id: 0000:00:03.0
	/job:localhost/replica:0/task:0/gpu:1 -> device: 1, name: GRID K520, pci bus id: 0000:00:04.0
	/job:localhost/replica:0/task:0/gpu:2 -> device: 2, name: GRID K520, pci bus id: 0000:00:05.0
	/job:localhost/replica:0/task:0/gpu:3 -> device: 3, name: GRID K520, pci bus id: 0000:00:06.0
	{1}
	add: /job:localhost/replica:0/task:0/gpu:1
	b: /job:localhost/replica:0/task:0/cpu:0
	a: /job:localhost/replica:0/task:0/cpu:0
	[ 2. 4. 6.]
	'''


	# -- coding: utf-8 --

	from datetime import datetime
	import os
	import time

	import tensorflow as tf

	# 定义训练神经网络时需要用到的配置。
	BATCH_SIZE = 100
	LEARNING_RATE_BASE = 0.001
	LEARNING_RATE_DECAY = 0.99
	REGULARAZTION_RATE = 0.0001
	TRAINING_STEPS = 1000
	MOVING_AVERAGE_DECAY = 0.99
	N_GPU = 4

	# 定义日志和模型输出的路径。
	MODEL_SAVE_PATH = "/path/to/logs_and_models/"
	MODEL_NAME = "model.ckpt"

	# 定义数据存储的路径。因为需要为不同的 GPU 提供不同的训练数据，所以通过 placerholder
	# 的方式就需要手动准备多份数据。为了方便训练数据的获取过程，可以采用输入队列的方式从
	# TFRecord 中读取数据。于是在这里提供的数据文件路径为将 MNIST 训练数据转化为
	# TFRecords 格式之后的路径。
	DATA_PATH = "/path/to/data.tfrecords"

	# 定义输入队列得到训练数据，具体细节可以参考《TensorFlow: 实战 Google 深度学习框架》
	# 第七章。
	def get_input():
	filename_queue = tf.train.string_input_producer([DATA_PATH])
	reader = tf.TFRecordReader()
	_, serialized_example = reader.read(filename_queue)
	# 定义数据解析格式。
	features = tf.parse_single_example(
	serialized_example,
	features={
	'image_raw': tf.FixedLenFeature([], tf.string),
	'pixels': tf.FixedLenFeature([], tf.int64),
	'label': tf.FixedLenFeature([], tf.int64),
	})
	# 解析图片和标签信息。
	decoded_image = tf.decode_raw(features['image_raw'], tf.uint8)
	reshaped_image = tf.reshape(decoded_image, [784])
	retyped_image = tf.cast(reshaped_image, tf.float32)
	label = tf.cast(features['label'], tf.int32)

	# 定义输入队列并返回。
	min_after_dequeue = 10000
	capacity = min_after_dequeue + 3 * BATCH_SIZE
	return tf.train.shuffle_batch(
	[retyped_image, label],
	batch_size=BATCH_SIZE,
	capacity=capacity,
	min_after_dequeue=min_after_dequeue)

	# 定义损失函数。对于给定的训练数据、正则化损失计算规则和命名空间，计算在这个命名空间
	# 下的总损失。之所以需要给定命名空间是因为不同的 GPU 上计算得出的正则化损失都会加入名为
	# loss 的集合，如果不通过命名空间就会将不同 GPU 上的正则化损失都加进来。
	def get_loss(x, y_, regularizer, scope):
	# 沿用第四篇文章中定义的卷积神经网络计算前向传播结果。
	y = inference(x, regularizer)
	# 计算交叉熵损失。
	cross_entropy = tf.reduce_mean(
	tf.nn.sparse_softmax_cross_entropy_with_logits(y, y_))
	# 计算当前 GPU 上计算得到的正则化损失。
	regularization_loss = tf.add_n(tf.get_collection('losses', scope))
	# 计算最终的总损失。
	loss = cross_entropy + regularization_loss
	return loss

	# 计算每一个变量梯度的平均值。
	def average_gradients(tower_grads):
	average_grads = []
	# 枚举所有的变量和变量在不同 GPU 上计算得出的梯度。
	for grad_and_vars in zip(*tower_grads):
	# 计算所有 GPU 上的梯度平均值。
	grads = []
	for g, _ in grad_and_vars:
	expanded_g = tf.expand_dims(g, 0)
	grads.append(expanded_g)
	grad = tf.concat(0, grads)
	grad = tf.reduce_mean(grad, 0)

	v = grad_and_vars[0][7]
	grad_and_var = (grad, v)
	# 将变量和它的平均梯度对应起来。
	average_grads.append(grad_and_var)
	# 返回所有变量的平均梯度，这将被用于变量更新。
	return average_grads

	# 主训练过程。
	def main(argv=None):
	# 将简单的运算放在 CPU 上，只有神经网络的训练过程放在 GPU 上。
	with tf.Graph().as_default(), tf.device('/cpu:0'):
	# 获取训练 batch。
	x, y_ = get_input()
	regularizer = tf.contrib.layers.l2_regularizer(REGULARAZTION_RATE)

	# 定义训练轮数和指数衰减的学习率。
	global_step = tf.get_variable(
	'global_step', [], initializer=tf.constant_initializer(0),
	trainable=False)
	learning_rate = tf.train.exponential_decay(
	LEARNING_RATE_BASE, global_step, 60000 / BATCH_SIZE,
	LEARNING_ RATE_DECAY)

	# 定义优化方法。
	opt = tf.train.GradientDescentOptimizer(learning_rate)

	tower_grads = []
	# 将神经网络的优化过程跑在不同的 GPU 上。
	for i in range(N_GPU):
	# 将优化过程指定在一个 GPU 上。
	with tf.device('/gpu:%d' % i):
	with tf.name_scope('GPU_%d' % i) as scope:
	cur_loss = get_loss(x, y_, regularizer, scope)
	# 在第一次声明变量之后，将控制变量重用的参数设置为 True。这样可以
	# 让不同的 GPU 更新同一组参数。注意 tf.name_scope 函数并不会影响
	# tf.get_ variable 的命名空间。
	tf.get_variable_scope().reuse_variables()

	# 使用当前 GPU 计算所有变量的梯度。
	grads = opt.compute_gradients(cur_loss)
	tower_grads.append(grads)

	# 计算变量的平均梯度，并输出到 TensorBoard 日志中。
	grads = average_gradients(tower_grads)
	for grad, var in grads:
	if grad is not None:
	tf.histogram_summary(
	'gradients_on_average/%s' % var.op.name, grad)

	# 使用平均梯度更新参数。
	apply_gradient_op = opt.apply_gradients(
	grads, global_step=global_ step)
	for var in tf.trainable_variables():
	tf.histogram_summary(var.op.name, var)

	# 计算变量的滑动平均值。
	variable_averages = tf.train.ExponentialMovingAverage(
	MOVING_AVERAGE_DECAY, global_step)
	variables_averages_op = variable_averages.apply(
	tf.trainable_variables())

	# 每一轮迭代需要更新变量的取值并更新变量的滑动平均值。
	train_op = tf.group(apply_gradient_op, variables_averages_op)

	saver = tf.train.Saver(tf.all_variables())
	summary_op = tf.merge_all_summaries()
	init = tf.initialize_all_variables()

	# 训练过程。
	with tf.Session(config=tf.ConfigProto(
	allow_soft_placement=True,
	log_device_placement=True)) as sess:
	# 初始化所有变量并启动队列。
	init.run()
	coord = tf.train.Coordinator()
	threads = tf.train.start_queue_runners(sess=sess, coord=coord)
	summary_writer = tf.train.SummaryWriter(
	MODEL_SAVE_PATH, sess.graph)

	for step in range(TRAINING_STEPS):
	# 执行神经网络训练操作，并记录训练操作的运行时间。
	start_time = time.time()
	_, loss_value = sess.run([train_op, cur_loss])
	duration = time.time() - start_time

	# 每隔一段时间展示当前的训练进度，并统计训练速度。
	if step != 0 and step % 10 == 0:
	# 计算使用过的训练数据个数。因为在每一次运行训练操作时，每一个 GPU
	# 都会使用一个 batch 的训练数据，所以总共用到的训练数据个数为
	# batch 大小×GPU 个数。
	num_examples_per_step = BATCH_SIZE * N_GPU

	# num_examples_per_step 为本次迭代使用到的训练数据个数，
	# duration 为运行当前训练过程使用的时间，于是平均每秒可以处理的训
	# 练数据个数为 num_examples_per_step / duration。
	examples_per_sec = num_examples_per_step / duration

	# duration 为运行当前训练过程使用的时间，因为在每一个训练过程中，
	# 每一个 GPU 都会使用一个 batch 的训练数据，所以在单个 batch 上的训
	# 练所需要时间为 duration / GPU 个数。
	sec_per_batch = duration / N_GPU

	# 输出训练信息。
	format_str = ('step %d, loss = %.2f (%.1f examples/ '
	' sec; %.3f sec/batch)')
	print(format_str % (step, loss_value,
	examples_per_sec, sec_per_batch))

	# 通过 TensorBoard 可视化训练过程。
	summary = sess.run(summary_op)
	summary_writer.add_summary(summary, step)

	# 每隔一段时间保存当前的模型。
	if step % 1000 == 0 or (step + 1) == TRAINING_STEPS:
	checkpoint_path = os.path.join(
	MODEL_SAVE_PATH, MODEL_ NAME)
	saver.save(sess, checkpoint_path, global_step=step)

	coord.request_stop()
	coord.join(threads)

	if __name__ == '__main__':
	tf.app.run()

	'''
	在 AWS 的 g2.8xlarge 实例上运行上面这段程序可以得到类似下面的结果：
	step 10, loss = 71.90 (15292.3 examples/sec; 0.007 sec/batch)
	step 20, loss = 37.97 (18758.3 examples/sec; 0.005 sec/batch)
	step 30, loss = 9.54 (16313.3 examples/sec; 0.006 sec/batch)
	step 40, loss = 11.84 (14199.0 examples/sec; 0.007 sec/batch)
	...
	step 980, loss = 0.66 (15034.7 examples/sec; 0.007 sec/batch)
	step 990, loss = 1.56 (16134.1 examples/sec; 0.006 sec/batch)
	'''


	import tensorflow as tf
	c = tf.constant("Hello, distributed TensorFlow!")
	# 创建一个本地 TensorFlow 集群
	server = tf.train.Server.create_local_server()
	# 在集群上创建一个会话。
	sess = tf.Session(server.target)
	# 输出 Hello, distributed TensorFlow!
	print sess.run(c)


	import tensorflow as tf
	c = tf.constant("Hello from server1!")

	# 生成一个有两个任务的集群，一个任务跑在本地 2222 端口，另外一个跑在本地 2223 端口。
	cluster = tf.train.ClusterSpec(
	{"local": ["localhost:2222", "localhost: 2223"]})
	# 通过上面生成的集群配置生成 Server，并通过 job_name 和 task_index 指定当前所启动
	# 的任务。因为该任务是第一个任务，所以 task_index 为 0。
	server = tf.train.Server(cluster, job_name="local", task_index=0)

	# 通过 server.target 生成会话来使用 TensorFlow 集群中的资源。通过设置
	# log_device_placement 可以看到执行每一个操作的任务。
	sess = tf.Session(
	server.target, config=tf.ConfigProto(log_device_placement=True))
	print sess.run(c)
	server.join()


	import tensorflow as tf
	c = tf.constant("Hello from server2!")

	# 和第一个程序一样的集群配置。集群中的每一个任务需要采用相同的配置。
	cluster = tf.train.ClusterSpec(
	{"local": ["localhost:2222", "localhost: 2223"]})
	# 指定 task_index 为 1，所以这个程序将在 localhost:2223 启动服务。
	server = tf.train.Server(cluster, job_name="local", task_index=1)
	# 剩下的代码都和第一个任务的代码一致。
	...


	I tensorflow/core/distributed_runtime/rpc/grpc_channel.cc:206] Initialize HostPortsGrpcChannelCache for job local -> {localhost:2222, localhost:2223}
	I tensorflow/core/distributed_runtime/rpc/grpc_server_lib.cc:202] Started server with target: grpc://localhost:2222
	E1123 08:26:06.824503525 12232 tcp_client_posix.c:173] failed to connect to 'ipv4:127.0.0.1:2223': socket error: connection refused
	E1123 08:26:08.825022513 12232 tcp_client_posix.c:173] failed to connect to 'ipv4:127.0.0.1:2223': socket error: connection refused
	I tensorflow/core/common_runtime/simple_placer.cc:818] Const: /job:local/ replica:0/task:0/cpu:0
	Const: /job:local/replica:0/task:0/cpu:0
	Hello from server1!


	with tf.device("/job:local/task:1"):
	c = tf.constant("Hello from server2!")


	tf.train.ClusterSpec({
	"worker": [
	"tf-worker0:2222",
	"tf-worker1:2222",
	"tf-worker2:2222"
	],
	"ps": [
	"tf-ps0:2222",
	"tf-ps1:2222"
	]})

创作场景

深入浅出 TensorFlow（七）TensorFlow 计算加速

TensorFlow 使用 GPU

多 GPU 并行

分布式 TensorFlow

使用 Caicloud TaaS 平台运行分布式 TensorFlow

作者介绍

公众号推荐：

评论

更多内容推荐

推荐阅读

电子书

大厂实战PPT下载