tensorflow分布式训练
Posted lixiaolun
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了tensorflow分布式训练相关的知识,希望对你有一定的参考价值。
转自:https://blog.csdn.net/daydayup_668819/article/details/80006229
一、单机单卡
单机单卡是最普通的情况,当然也是最简单的,示例代码如下:
#coding=utf-8 #单机单卡 #对于单机单卡,可以把参数和计算都定义在gpu上,不过如果参数模型比较大,显存不足等情况,就得放在cpu上 import tensorflow as tf with tf.device(‘/cpu:0‘):#也可以放在gpu上 w=tf.get_variable(‘w‘,(2,2),tf.float32,initializer=tf.constant_initializer(2)) b=tf.get_variable(‘b‘,(2,2),tf.float32,initializer=tf.constant_initializer(5)) with tf.device(‘/gpu:0‘): addwb=w+b mutwb=w*b ini=tf.initialize_all_variables() with tf.Session() as sess: sess.run(ini) np1,np2=sess.run([addwb,mutwb]) print np1 print np2
二、单机多卡
单机多卡,只要用device直接指定设备,就可以进行训练,SGD采用各个卡的平均值,示例代码如下:
#coding=utf-8 #单机多卡: #一般采用共享操作定义在cpu上,然后并行操作定义在各自的gpu上,比如对于深度学习来说,一般把参数定义、参数梯度更新统一放在cpu上 #各个gpu通过各自计算各自batch 数据的梯度值,然后统一传到cpu上,由cpu计算求取平均值,cpu更新参数。 #具体的深度学习多卡训练代码,请参考: https://github.com/tensorflow/models/blob/master/research/inception/inception/inception_distributed_train.py import tensorflow as tf with tf.device(‘/cpu:0‘): w=tf.get_variable(‘w‘,(2,2),tf.float32,initializer=tf.constant_initializer(2)) b=tf.get_variable(‘b‘,(2,2),tf.float32,initializer=tf.constant_initializer(5)) with tf.device(‘/gpu:0‘): addwb=w+b with tf.device(‘/gpu:1‘): mutwb=w*b ini=tf.initialize_all_variables() with tf.Session() as sess: sess.run(ini) while 1: print sess.run([addwb,mutwb])
模型训练时:
● 在每个GPU上放置个体模型的复制品。
● 在GPU上分批次。
● 等待所有GPU处理完一批数据,同步更新模型参数。
每个塔计算批次的一部分梯度,将多个塔上的梯度求和并计算平均值,以此提供一个单一的更新的变量并存储在CPU上。
三、多机多卡
1. Cluster、Job、task介绍
Task可看作是主机上的一个进程,在大多数情况下,一个机器上只运行一个Task。
在分布式深度学习框架中,一般把Job划分为Parameter Server和Worker。ps是管理参数的存储和更新工作,worker是来运行ops。如果参数的数量太大,一台机器处理不了,这就要需要多个Tasks。
Cluster,就是用的集群系统。
三者关系:
多个Task组成Job,多个Job组成Cluster;即:Job是Task的集合,Cluster是Job的集合。
TensorFlow中的集群(cluster)指的是一系列能够对TensorFlow中的图(graph)进行分布式计算的任务(task)。每个任务是同服务(server)相关联的。TensorFlow中的服务会包含一个用于创建session的主节点和一个用于图运算的工作节点。另外, TensorFlow中的集群可以拆分成一个或多个作业(job), 每个作业可以包含一个或多个任务。下图为对集群内关系的理解。
2. 同步SGD与异步SGD
所谓的同步更新指的是:各个用于并行计算的电脑,计算完各自的batch 后,求取梯度值,把梯度值统一送到ps服务机器中,由ps服务机器求取梯度平均值,更新ps服务器上的参数。
如下图所示,可以看成有四台电脑,第一台电脑用于存储参数、共享参数、共享计算,可以简单的理解成内存、计算共享专用的区域,也就是ps job;另外三台电脑用于并行计算的,也就是worker task。
这种计算方法存在的缺陷是:每一轮的梯度更新,都要等到A、B、C三台电脑都计算完毕后,才能更新参数,也就是迭代更新速度取决与A、B、C三台中,最慢的那一台电脑。
所谓的异步更新指的是:ps服务器收到只要收到一台机器的梯度值,就直接进行参数更新,无需等待其它机器。这种迭代方法比较不稳定,收敛曲线震动比较厉害,因为当A机器计算完更新了ps中的参数,可能B机器还是在用上一次迭代的旧版参数值。
3. 代码编写
1) 定义集群
创建集群的必要条件是为每个任务启动一个服务。这些任务可以运行在不同的机器上,也可以在同一台机器上启动多个任务(比如在本地多个不同的GPU上运行)。每个任务会做如下的两步工作:
a. 创建一个 tf.train.ClusterSpec 用于对集群中的所有任务进行描述,该描述内容对于所有任务应该是相同的。
b. 创建一个tf.train.Server 并将tf.train.ClusterSpec 中的参数传入构造函数,并将作业的名称和当前任务的编号写入本地任务中。
比如假设上面的图所示,我们有四台电脑,四台电脑的名字假设为:A、B、C、D,那么集群可以定义如下:
#coding=utf-8 #多台机器,每台机器有一个显卡、或者多个显卡,这种训练叫做分布式训练 import tensorflow as tf #现在假设我们有A、B、C、D四台机器,首先需要在各台机器上写一份代码,并跑起来,各机器上的代码内容大部分相同 # 除了开始定义的时候,需要各自指定该台机器的task之外。以机器A为例子,A机器上的代码如下: cluster=tf.train.ClusterSpec({ "worker": [ "A_IP:2222",#格式 IP地址:端口号,第一台机器A的IP地址 ,在代码中需要用这台机器计算的时候,就要定义:/job:worker/task:0 "B_IP:1234"#第二台机器的IP地址 /job:worker/task:1 "C_IP:2222"#第三台机器的IP地址 /job:worker/task:2 ], "ps": [ "D_IP:2222",#第四台机器的IP地址 对应到代码块:/job:ps/task:0 ]})
需要写四分代码,这四分代码文件大部分相同,但是有几行代码是各不相同的。
2) 在各台机器上定义server
将ClusterSpec当作参数传入到 tf.train.Server()中,同时指定此Task的Job_name和task_index。比如A机器上的代码server要定义如下:
server=tf.train.Server(cluster,job_name=‘worker‘,task_index=0)#找到‘worker’名字下的,task0,也就是机器A
tf.trian.Server干了些什么呢?
首先,一个tf.train.Server包含了本地设备(GPUs,CPUs)的集合,可以连接到其它task的ip:port(存储在cluster中), 还有一个session target用来执行分布操作。还有就是,它创建了一个服务器,监听port端口,如果有数据传过来,它就会在本地执行(启动session target,调用本地设备执行运算),然后结果返回给调用者。
3) 指定分布式device
with tf.device(‘/job:ps/task:0‘):#参数定义在机器D上 w=tf.get_variable(‘w‘,(2,2),tf.float32,initializer=tf.constant_initializer(2)) b=tf.get_variable(‘b‘,(2,2),tf.float32,initializer=tf.constant_initializer(5)) with tf.device(‘/job:worker/task:0/cpu:0‘):#在机器A cpu上运行 addwb=w+b with tf.device(‘/job:worker/task:1/cpu:0‘):#在机器B cpu上运行 mutwb=w*b with tf.device(‘/job:worker/task:2/cpu:0‘):#在机器C cpu上运行 divwb=w/b
在深度学习训练中,一般图的计算,对于每个worker task来说,都是相同的,所以我们会把所有图计算、变量定义等代码,都写到下面这个语句下:
with tf.device(tf.train.replica_device_setter(worker_device=‘/job:worker/task:index:%d‘,cluster=cluster)):
函数replica_deviec_setter会自动把变量参数定义部分定义到ps服务中(如果ps有多个任务,那么自动分配)。下面举个例子,假设现在有两台机器A、B,A用于计算服务,B用于参数服务,那么代码如下:
#coding=utf-8 #上面是因为worker计算内容各不相同,不过在深度学习中,一般每个worker的计算内容都是一样的, # 都是计算神经网络的每个batch前向传导,所以一般代码是重用的 import tensorflow as tf #现在假设我们有A、B台机器,首先需要在各台机器上写一份代码,并跑起来,各机器上的代码内容大部分相同 # 除了开始定义的时候,需要各自指定该台机器的task之外。以机器A为例子,A机器上的代码如下: cluster=tf.train.ClusterSpec({ "worker": [ "192.168.11.105:1234",#格式 IP地址:端口号,第一台机器A的IP地址 ,在代码中需要用这台机器计算的时候,就要定义:/job:worker/task:0 ], "ps": [ "192.168.11.130:2223"#第四台机器的IP地址 对应到代码块:/job:ps/task:0 ]}) #不同的机器,下面这一行代码各不相同,server可以根据job_name、task_index两个参数,查找到集群cluster中对应的机器 isps=False if isps: server=tf.train.Server(cluster,job_name=‘ps‘,task_index=0)#找到‘worker’名字下的,task0,也就是机器A server.join() else: server=tf.train.Server(cluster,job_name=‘worker‘,task_index=0)#找到‘worker’名字下的,task0,也就是机器A with tf.device(tf.train.replica_device_setter(worker_device=‘/job:worker/task:0‘,cluster=cluster)): w=tf.get_variable(‘w‘,(2,2),tf.float32,initializer=tf.constant_initializer(2)) b=tf.get_variable(‘b‘,(2,2),tf.float32,initializer=tf.constant_initializer(5)) addwb=w+b mutwb=w*b divwb=w/b saver = tf.train.Saver() summary_op = tf.merge_all_summaries() init_op = tf.initialize_all_variables() sv = tf.train.Supervisor(init_op=init_op, summary_op=summary_op, saver=saver) with sv.managed_session(server.target) as sess: while 1: print sess.run([addwb,mutwb,divwb])
把该代码在机器A上运行,你会发现,程序会进入等候状态,等候用于ps参数服务的机器启动,才会运行。因此接着在机器B上运行如下代码:
#coding=utf-8 #上面是因为worker计算内容各不相同,不过再深度学习中,一般每个worker的计算内容是一样的, # 以为都是计算神经网络的每个batch 前向传导,所以一般代码是重用的 #coding=utf-8 #多台机器,每台机器有一个显卡、或者多个显卡,这种训练叫做分布式训练 import tensorflow as tf #现在假设我们有A、B、C、D四台机器,首先需要在各台机器上写一份代码,并跑起来,各机器上的代码内容大部分相同 # ,除了开始定义的时候,需要各自指定该台机器的task之外。以机器A为例子,A机器上的代码如下: cluster=tf.train.ClusterSpec({ "worker": [ "192.168.11.105:1234",#格式 IP地址:端口号,第一台机器A的IP地址 ,在代码中需要用这台机器计算的时候,就要定义:/job:worker/task:0 ], "ps": [ "192.168.11.130:2223"#第四台机器的IP地址 对应到代码块:/job:ps/task:0 ]}) #不同的机器,下面这一行代码各不相同,server可以根据job_name、task_index两个参数,查找到集群cluster中对应的机器 isps=True if isps: server=tf.train.Server(cluster,job_name=‘ps‘,task_index=0)#找到‘worker’名字下的,task0,也就是机器A server.join() else: server=tf.train.Server(cluster,job_name=‘worker‘,task_index=0)#找到‘worker’名字下的,task0,也就是机器A with tf.device(tf.train.replica_device_setter(worker_device=‘/job:worker/task:0‘,cluster=cluster)): w=tf.get_variable(‘w‘,(2,2),tf.float32,initializer=tf.constant_initializer(2)) b=tf.get_variable(‘b‘,(2,2),tf.float32,initializer=tf.constant_initializer(5)) addwb=w+b mutwb=w*b divwb=w/b saver = tf.train.Saver() summary_op = tf.merge_all_summaries() init_op = tf.initialize_all_variables() sv = tf.train.Supervisor(init_op=init_op, summary_op=summary_op, saver=saver) with sv.managed_session(server.target) as sess: while 1: print sess.run([addwb,mutwb,divwb])
参考资料:
https://github.com/tensorflow/models/tree/master/research/inception
https://blog.csdn.net/hjimce/article/details/61197190
以上是关于tensorflow分布式训练的主要内容,如果未能解决你的问题,请参考以下文章