Tensorflow cpu分布式原理和运行示例

小鱼儿 2021-09-28 18:24 321阅读 0赞

## 1、TF分布式原理 ##

TF的实现分为了单机实现和分布式实现，在分布式实现中，需要实现的是对client，master，worker process不在同一台机器上时的支持。数据量很大的情况下，单机跑深度学习程序，过于耗时，所以需要TensorFlow分布式并行。

![4.png_version_1_modificationDate_1516153156797_api_v2][]

## 2、TF分布式模式 ##

In-graph 模式

将模型的计算图的不同部分放在不同的机器上执行。

在大数据训练的情况下， 不推荐使用这种模式。

Between-graph 模式

数据并行，每台机器使用完全相同的计算图。

这种模式的优点是不用训练数据的分发了， 尤其是在数据量在TB级的时候， 节省了大量的时间，所以大数据深度学习还是推荐使用Between-graph模式。

## 3、同步更新和异步更新 ##

in-graph模式和between-graph模式都支持同步和异步更新。

在数据量小的情况下， 各个节点的计算能力比较均衡的情况下， 推荐使用同步模式；数据量很大，各个机器的计算性能参差不齐的情况下，推荐使用异步的方式。

## 4、Single-Device Execution ##

构建好图后，使用拓扑算法来决定执行哪一个节点，即对每个节点使用一个计数，值表示所依赖的未完成的节点数目，当一个节点的运算完成时，将依赖该节点的所有节点的计数减一。如果节点的计数为0，将其放入准备队列待执行。

## 5、单机多GPU训练 ##

单机多GPU的训练过程：

1.  假设你的机器上有3个GPU;
2.  在单机单GPU的训练中，数据是一个batch一个batch的训练。 在单机多GPU中，数据一次处理3个batch(假设是3个GPU训练）， 每个GPU处理一个batch的数据计算。
3.  变量，或者说参数，保存在CPU上
4.  刚开始的时候数据由CPU分发给3个GPU， 在GPU上完成了计算，得到每个batch要更新的梯度。
5.  然后在CPU上收集完了3个GPU上的要更新的梯度， 计算一下平均梯度，然后更新参数。
6.  然后继续循环这个过程。

通过这个过程，处理的速度取决于最慢的那个GPU的速度。如果3个GPU的处理速度差不多的话， 处理速度就相当于单机单GPU的速度的3倍减去数据在CPU和GPU之间传输的开销，实际的效率提升看CPU和GPU之间数据的速度和处理数据的大小。

## 6、Multi-Device Execution ##

当系统到了分布式情况下时，事情就变得复杂了很多，还好前述调度用了现有的框架。那么对于TF来说，剩下的事情就是：

决定运算在哪个设备上运行 管理设备之间的数据传递

## 7、分布式多机多GPU训练 ##

随着设计的模型越来越复杂，模型参数越来越多，越来越大， 大到什么程度？多到什么程度？ 多参数的个数上百亿个， 训练的数据多到按TB级别来衡量。大家知道每次计算一轮，都要计算梯度，更新参数。 当参数的量级上升到百亿量级甚至更大之后， 参数的更新的性能都是问题。 如果是单机16个GPU， 一个step最多也是处理16个batch， 这对于上TB级别的数据来说，不知道要训练到什么时候。于是就有了分布式的深度学习训练方法，或者说框架。

## 8、参数服务器 ##

在介绍tensorflow的分布式训练之前，先说下参数服务器的概念。 前面说道， 当你的模型越来越大， 模型的参数越来越多，多到模型参数的更新，一台机器的性能都不够的时候， 很自然的我们就会想到把参数分开放到不同的机器去存储和更新。 因为碰到上面提到的那些问题， 所有参数服务器就被单独拧出来， 于是就有了参数服务器的概念。 参数服务器可以是多台机器组成的集群， 这个就有点类似分布式的存储架构了， 涉及到数据的同步，一致性等等， 一般是key-value的形式，可以理解为一个分布式的key-value内存数据库，然后再加上一些参数更新的操作。 详细的细节可以去google一下， 这里就不详细说了。 反正就是当性能不够的时候， 几百亿的参数分散到不同的机器上去保存和更新，解决参数存储和更新的性能问题。 借用上面的小明算题的例子，小明觉得自己算加法都算不过来了， 于是就叫了10个小明过来一起帮忙算。

## 9、gRPC (google remote procedure call) ##

TensorFlow分布式并行基于gRPC通信框架，其中包括一个master创建Session，还有多个worker负责执行计算图中的任务。

gRPC首先是一个RPC，即远程过程调用,通俗的解释是：假设你在本机上执行一段代码`num=add(a,b)`，它调用了一个过程 call，然后返回了一个值num，你感觉这段代码只是在本机上执行的, 但实际情况是,本机上的add方法是将参数打包发送给服务器,然后服务器运行服务器端的add方法,返回的结果再将数据打包返回给客户端.

## 10、结构 ##

Cluster是Job的集合，Job是Task的集合。

即：一个Cluster可以切分多个Job，一个Job指一类特定的任务，每个Job包含多个Task，比如parameter server(ps)、worker，在大多数情况下,一个机器上只运行一个Task.

在分布式深度学习框架中,我们一般把Job划分为Parameter Server和Worker:

*  Parameter Job是管理参数的存储和更新工作.
 *  Worker Job是来运行ops.

如果参数的数量太大,一台机器处理不了,这就要需要多个Tasks.

## 11、普通代码改为分布式步骤 ##

分布式代码其实就是多了一个框架放在普通代码外面。

1、通过定义tf.app.flags标记接收命令行传递的参数

2、在main函数中先创建Cluster对象和server对象，parameter节点等待worker节点工作

3、在worker节点中定义神经网络模型：把普通代码拷贝过来，稍作修改，比如参数改为从flag获取、增加global\_step

4、初始化本地参数，创建分布式训练的监督器Supervisor

5、在循环中执行具体的训练过程，每次读取一个batch的数据来训练，达到最大训练步数后停止

## 12、cnn手写体识别 ##

# encoding:utf-8
    import math
    import tempfile
    import time
    import tensorflow as tf
    from tensorflow.examples.tutorials.mnist import input_data
     
    flags = tf.app.flags
    IMAGE_PIXELS = 28
    # 定义默认训练参数和数据路径
    flags.DEFINE_string('data_dir', '/opt/pythonenv/Tensorflow入门/dataset/MNIST_data/', 'Directory  for storing mnist data')
    flags.DEFINE_integer('hidden_units', 100, 'Number of units in the hidden layer of the NN')
    flags.DEFINE_integer('train_steps', 10000, 'Number of training steps to perform')
    flags.DEFINE_integer('batch_size', 100, 'Training batch size ')
    flags.DEFINE_float('learning_rate', 0.01, 'Learning rate')
    # 定义分布式参数
    # 参数服务器parameter server节点
    flags.DEFINE_string('ps_hosts', '192.168.1.219:22221', 'Comma-separated list of hostname:port pairs')
    # 两个worker节点
    flags.DEFINE_string('worker_hosts', '192.168.1.219:22222,192.168.1.219:22223',
                        'Comma-separated list of hostname:port pairs')
    # 设置job name参数
    flags.DEFINE_string('job_name', None, 'job name: worker or ps')
    # 设置任务的索引
    flags.DEFINE_integer('task_index', None, 'Index of task within the job')
    # 选择异步并行，同步并行
    flags.DEFINE_integer("issync", None, "是否采用分布式的同步模式，1表示同步模式，0表示异步模式")
     
    FLAGS = flags.FLAGS
     
     
    def main(unused_argv):
        mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True)
     
        if FLAGS.job_name is None or FLAGS.job_name == '':
            raise ValueError('Must specify an explicit job_name !')
        else:
            print( 'job_name : %s' % FLAGS.job_name)
        if FLAGS.task_index is None or FLAGS.task_index == '':
            raise ValueError('Must specify an explicit task_index!')
        else:
            print( 'task_index : %d' % FLAGS.task_index)
     
        ps_spec = FLAGS.ps_hosts.split(',')
        worker_spec = FLAGS.worker_hosts.split(',')
     
        # 创建集群
        num_worker = len(worker_spec)
        cluster = tf.train.ClusterSpec({'ps': ps_spec, 'worker': worker_spec})
        server = tf.train.Server(cluster, job_name=FLAGS.job_name, task_index=FLAGS.task_index)
        if FLAGS.job_name == 'ps':
            server.join()
     
        is_chief = (FLAGS.task_index == 0)
        # worker_device = '/job:worker/task%d/cpu:0' % FLAGS.task_index
        with tf.device(tf.train.replica_device_setter(
                cluster=cluster
        )):
            global_step = tf.Variable(0, name='global_step', trainable=False)  # 创建纪录全局训练步数变量
     
            hid_w = tf.Variable(tf.truncated_normal([IMAGE_PIXELS * IMAGE_PIXELS, FLAGS.hidden_units],
                                                    stddev=1.0 / IMAGE_PIXELS), name='hid_w')
            hid_b = tf.Variable(tf.zeros([FLAGS.hidden_units]), name='hid_b')
     
            sm_w = tf.Variable(tf.truncated_normal([FLAGS.hidden_units, 10],
                                                   stddev=1.0 / math.sqrt(FLAGS.hidden_units)), name='sm_w')
            sm_b = tf.Variable(tf.zeros([10]), name='sm_b')
     
            x = tf.placeholder(tf.float32, [None, IMAGE_PIXELS * IMAGE_PIXELS])
            y_ = tf.placeholder(tf.float32, [None, 10])
     
            hid_lin = tf.nn.xw_plus_b(x, hid_w, hid_b)
            hid = tf.nn.relu(hid_lin)
     
            y = tf.nn.softmax(tf.nn.xw_plus_b(hid, sm_w, sm_b))
            cross_entropy = -tf.reduce_sum(y_ * tf.log(tf.clip_by_value(y, 1e-10, 1.0)))
     
            opt = tf.train.AdamOptimizer(FLAGS.learning_rate)
     
            train_step = opt.minimize(cross_entropy, global_step=global_step)
            # 生成本地的参数初始化操作init_op
            init_op = tf.global_variables_initializer()
            train_dir = tempfile.mkdtemp()
            sv = tf.train.Supervisor(is_chief=is_chief, logdir=train_dir, init_op=init_op, recovery_wait_secs=1,
                                     global_step=global_step)
     
            if is_chief:
                print( 'Worker %d: Initailizing session...' % FLAGS.task_index)
            else:
                print('Worker %d: Waiting for session to be initaialized...' % FLAGS.task_index)
            sess = sv.prepare_or_wait_for_session(server.target)
            print( 'Worker %d: Session initialization  complete.' % FLAGS.task_index)
     
            time_begin = time.time()
            print( 'Traing begins @ %f' % time_begin)
     
            local_step = 0
            while True:
                batch_xs, batch_ys = mnist.train.next_batch(FLAGS.batch_size)
                train_feed = {x: batch_xs, y_: batch_ys}
     
                _, step = sess.run([train_step, global_step], feed_dict=train_feed)
                local_step += 1
     
                now = time.time()
                print( '%f: Worker %d: traing step %d dome (global step:%d)' % (now, FLAGS.task_index, local_step, step))
     
                if step >= FLAGS.train_steps:
                    break
     
            time_end = time.time()
            print( 'Training ends @ %f' % time_end)
            train_time = time_end - time_begin
            print( 'Training elapsed time:%f s' % train_time)
     
            val_feed = {x: mnist.validation.images, y_: mnist.validation.labels}
            val_xent = sess.run(cross_entropy, feed_dict=val_feed)
            print( 'After %d training step(s), validation cross entropy = %g' % (FLAGS.train_steps, val_xent))
        sess.close()
     
    if __name__ == '__main__':
        tf.app.run()

在单机上运行：

打开三个终端，先切换到tensorflow1.4环境下（执行 source /opt/pythonenv/tensorflow1.4/bin/activate），然后分别执行下面三行语句：
     
    python distributed_minst.py --job_name=ps --task_index=0 --ps_hosts=192.168.1.219:22224 --worker_hosts=192.168.1.219:22225,192.168.1.219:22226
     
    python distributed_minst.py --job_name=worker --task_index=0 --ps_hosts=192.168.1.219:22224 --worker_hosts=192.168.1.219:22225,192.168.1.219:22226
     
    python distributed_minst.py --job_name=worker --task_index=1 --ps_hosts=192.168.1.219:22224 --worker_hosts=192.168.1.219:22225,192.168.1.219:22226

如果是要在多台机器运行，修改ps\_hosts和worker\_hosts参数即可。

## 13、lstm手写数字识别 ##

# encoding:utf-8
    import math
    import tempfile
    import time
    import tensorflow as tf
    from tensorflow.examples.tutorials.mnist import input_data
     
    flags = tf.app.flags
    IMAGE_PIXELS = 28
    tf.set_random_seed(35)
     
     
    # hyperparameters设置模型参数，该参数是确定模型训练方式的，定义默认训练参数和数据路径
    flags.DEFINE_string('data_dir', '/opt/pythonenv/Tensorflow入门/dataset/MNIST_data/', 'Directory  for storing mnist data')
    flags.DEFINE_integer('n_hidden_units', 128, 'Number of units in the hidden layer of the NN')
    flags.DEFINE_integer('train_steps', 10000, 'Number of training steps to perform')
    flags.DEFINE_integer('batch_size', 128, 'Training batch size ')
    flags.DEFINE_float('lr', 0.001, 'Learning rate')
     
    flags.DEFINE_integer('n_inputs', 28, 'MNIST data input (img shape: 28*28)28维 ')
    flags.DEFINE_integer('n_steps', 28, 'time steps28行，每次读一行数据训练 ')
    flags.DEFINE_integer('n_classes', 10, 'MNIST classes (0-9 digits) ')
     
     
    # 定义分布式参数
    # 参数服务器parameter server节点
    flags.DEFINE_string('ps_hosts', '192.168.1.219:22221', 'Comma-separated list of hostname:port pairs')
    # 两个worker节点
    flags.DEFINE_string('worker_hosts', '192.168.1.219:22222,192.168.1.219:22223',
                        'Comma-separated list of hostname:port pairs')
    # 设置job name参数
    flags.DEFINE_string('job_name', None, 'job name: worker or ps')
    # 设置任务的索引
    flags.DEFINE_integer('task_index', None, 'Index of task within the job')
    # 选择异步并行，同步并行
    flags.DEFINE_integer("issync", None, "是否采用分布式的同步模式，1表示同步模式，0表示异步模式")
     
    FLAGS = flags.FLAGS
     
     
    def main(unused_argv):
       mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True)
        
       if FLAGS.job_name is None or FLAGS.job_name == '':
          raise ValueError('Must specify an explicit job_name !')
       else:
          print( 'job_name : %s' % FLAGS.job_name)
       if FLAGS.task_index is None or FLAGS.task_index == '':
          raise ValueError('Must specify an explicit task_index!')
       else:
          print( 'task_index : %d' % FLAGS.task_index)
     
       ps_spec = FLAGS.ps_hosts.split(',')
       worker_spec = FLAGS.worker_hosts.split(',')
     
       # 创建集群
       num_worker = len(worker_spec)
       cluster = tf.train.ClusterSpec({'ps': ps_spec, 'worker': worker_spec})
       server = tf.train.Server(cluster, job_name=FLAGS.job_name, task_index=FLAGS.task_index)
       if FLAGS.job_name == 'ps':
          server.join()
     
       is_chief = (FLAGS.task_index == 0)
       # worker_device = '/job:worker/task%d/cpu:0' % FLAGS.task_index
       with tf.device(tf.train.replica_device_setter(
          cluster=cluster
          )):
          global_step = tf.Variable(0, name='global_step', trainable=False)  # 创建纪录全局训练步数变量
     
          # tf Graph input，placehoder定义变量，hold住变量，在执行节点时传入数据给该变量
          x = tf.placeholder(tf.float32, [None, FLAGS.n_steps, FLAGS.n_inputs])  # none表示输入图片张数不确定
          y = tf.placeholder(tf.float32, [None, FLAGS.n_classes])
          # 初始化参数，该参数在模型训练过程中会不断更新，最后达到最优
          weights = {
             # 生成维度是(28, 128)的标准正太分布，mean=0，stddev=1
             'in': tf.Variable(tf.random_normal([FLAGS.n_inputs, FLAGS.n_hidden_units])),
             # (128, 10)
             'out': tf.Variable(tf.random_normal([FLAGS.n_hidden_units, FLAGS.n_classes]))
          }
          biases = {
             # (128, )
             'in': tf.Variable(tf.constant(0.1, shape=[FLAGS.n_hidden_units, ])),
             # (10, )
             'out': tf.Variable(tf.constant(0.1, shape=[FLAGS.n_classes, ]))
          }
     
          def RNN(X, weights, biases):
             # 输入层
             ########################################
             # transpose the inputs shape from
             # (128 batch ,28 steps, 28 inputs) ==> (128 batch * 28 steps, 28 inputs)
             X = tf.reshape(X, [-1, FLAGS.n_inputs])
        
             # into hidden
             # X_in = (128 batch * 28 steps, 128 hidden)
              
             X_in = tf.matmul(X, weights['in']) + biases['in']
              
             # X_in ==> (128 batch, 28 steps, 128 hidden)
              
             X_in = tf.reshape(X_in, [-1, FLAGS.n_steps, FLAGS.n_hidden_units])
              
             # 隐藏层
             ##########################################
        
             # BasicLSTMCell定义单个cell
             cell = tf.contrib.rnn.BasicLSTMCell(FLAGS.n_hidden_units)
             init_state = cell.zero_state(FLAGS.batch_size, dtype=tf.float32)
             # dynamic_rnn函数将cell连成RNN 网络
             outputs, final_state = tf.nn.dynamic_rnn(cell, X_in, initial_state=init_state)
        
             # 输出层：输出logits
             #############################################
        
             results = tf.matmul(final_state[1], weights['out']) + biases['out']
             # shape = (128, 10),每张图片10个logit值
        
             return results
     
          # 损失函数是整个minibatch的目标类别和预测类别之间的交叉熵,
          pred = RNN(x, weights, biases)
          cross_entropy = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(logits=pred, labels=y))
          train_op = tf.train.AdamOptimizer(FLAGS.lr)#.minimize(cost)
     
          correct_pred = tf.equal(tf.argmax(pred, 1), tf.argmax(y, 1))
          accuracy = tf.reduce_mean(tf.cast(correct_pred, tf.float32))
     
          train_step = train_op.minimize(cross_entropy, global_step=global_step)
          # 生成本地的参数初始化操作init_op
          init_op = tf.global_variables_initializer()
          train_dir = tempfile.mkdtemp()
          sv = tf.train.Supervisor(is_chief=is_chief, logdir=train_dir, init_op=init_op, recovery_wait_secs=1,
             global_step=global_step)
           
          if is_chief:
             print( 'Worker %d: Initailizing session...' % FLAGS.task_index)
          else:
             print('Worker %d: Waiting for session to be initaialized...' % FLAGS.task_index)
          sess = sv.prepare_or_wait_for_session(server.target)
          print( 'Worker %d: Session initialization  complete.' % FLAGS.task_index)
     
          time_begin = time.time()
          print( 'Traing begins @ %f' % time_begin)
     
          local_step = 0
          while True:
             batch_xs, batch_ys = mnist.train.next_batch(FLAGS.batch_size)
             batch_xs = batch_xs.reshape([FLAGS.batch_size, FLAGS.n_steps, FLAGS.n_inputs])
              
             train_feed = {x: batch_xs, y: batch_ys}
              
             _, step = sess.run([train_step, global_step], feed_dict=train_feed)
             local_step += 1
              
             now = time.time()
             print( '%f: Worker %d: traing step %d done (global step:%d) ' % (now, FLAGS.task_index, local_step, step))
     
             if step >= FLAGS.train_steps:
                break
     
          time_end = time.time()
          print( 'Training ends @ %f' % time_end)
          train_time = time_end - time_begin
          print( 'Training elapsed time:%f s' % train_time)
     
          val_feed = {x: mnist.validation.images, y: mnist.validation.labels}
          val_xent = sess.run(cross_entropy, feed_dict=val_feed)
          print( 'After %d training step(s), validation cross entropy = %g' % (FLAGS.train_steps, val_xent))
       sess.close()
     
    if __name__ == '__main__':
       tf.app.run()

在单机上运行：

打开三个终端，先切换到tensorflow1.4环境下（执行 source /opt/pythonenv/tensorflow1.4/bin/activate），然后分别执行下面三行语句：
     
    python lstm_minst.py --job_name=ps --task_index=0 --ps_hosts=192.168.1.219:22224 --worker_hosts=192.168.1.219:22225,192.168.1.219:22226
     
    python lstm_minst.py --job_name=worker --task_index=0 --ps_hosts=192.168.1.219:22224 --worker_hosts=192.168.1.219:22225,192.168.1.219:22226
     
    python lstm_minst.py --job_name=worker --task_index=1 --ps_hosts=192.168.1.219:22224 --worker_hosts=192.168.1.219:22225,192.168.1.219:22226

[4.png_version_1_modificationDate_1516153156797_api_v2]: http://219.142.47.93/confluence/download/attachments/9773625/4.png?version=1&modificationDate=1516153156797&api=v2