【6.5.3】配置分布式TensorFlow
现在手头上的一个tensorflow的项目，一个节点8个GPU，epoch一次，预计两天时间，搞个50次epoch的话，实在有些受不了。索性把我们有GPU的节点都利用起来，就有了这个分布式tensorfolw的笔记。
一、分布式TensorFlow角色以及原理
1.1 集群涉及的几个基本概念
客户端(Client)
客户端是一个用于建立TensorFlow计算图并创立与集群进行交互的会话层tensorflow::Session 的程序。一般客户端是通过python或C++实现的。一个独立的客户端进程可以同时与多个TensorFlow的服务端相连 ，同时一个独立的服务端也可以与多个客户端相连。
集群(Cluster)
一个TensorFlow的集群里包含了一个或多个作业(job), 每一个作业又可以拆分成一个或多个任务(task)。集群的概念主要用与一个特定的高层次对象中，比如说训练神经网络，并行化操作多台机器等等。集群对象可以通过tf.train.ClusterSpec 来定义。 作业(Job)
一个作业可以拆封成多个具有相同目的的任务(task)，比如说，一个称之为ps(parameter server，参数服务器)的作业中的任务主要是保存和更新变量，而一个名为worker(工作)的作业一般是管理无状态且主要从事计算的任务。一个作业中的任务可以运行于不同的机器上，作业的角色也是灵活可变的，比如说称之为”worker”的作业可以保存一些状态。
任务(Task)
任务相当于是一个特定的TesnsorFlow服务端，其相当于一个独立的进程，该进程属于特定的作业并在作业中拥有对应的序号。 TensorFlow服务端(TensorFlow server) 。
1.2 tensorflow集群中的几个基本概念
要想配置TensorFlow为分布训练，首先需要了解TensorFlow中关于分布式的角色分配。
- ps：作为分布式训练的服务端，等到各个终端(supervisors)来连接。
- worker：在TensorFlow的代码注释中被称为supervisors，作为分布式训练的运算终端。
- chief supervisors：在众多运算终端中必须选中一个作为主要的运算终端。该终端是在运算终端中最先启动的，它的功能是合并各个终端运算后的学习参数，将其保存再写入。
每个具体角色网络标识都是唯一的，即分布在不同IP的机器上(或者同一主机但不同端口号)。
在实际运行中，各个角色的网络构建部分代码必须完全相同。三者的分工如下：
- 服务器端作为一个多方协调者，等待各个运算终端来连接。
- cheif supervisors会在启动时统一管理全局的学习参数，进行初始化或从模型载入。
- 其它的运算终端只是负责得到其相应的任务并进行计算，并不会保存检查点以及用于TensorBoard可视化的summary日志等任何参数信息。
二、分布部署TensorFlow的具体方法
配置过程中，首先创建一个server，在server中会将ps以及所有worker的ip端口准备好，接着使用tf.train.Supervisor中的managed_seesion来管理打开的session，session只负责运算，而通信协调的事情就都交给Supervisor来管理了。
三、使用TensorFlow实现分布式部署训练
下面开始实现一个分布式训练的网络模型，仍然以线性回归的模型作为原型，并将其改为分布式。使我们需要在本机通过3个端口来建立3个终端，分别是ps，两个worker。代码主要分为以下几部分：
3.1 为每个角色创建IP地址和端口，创建server
首先创建集群(cluster)， ClusterSpec的定义，需要把你要跑这个任务的所有的ps和worker 的节点的ip和端口的信息都包含进去， 所有的角色都要执行这段代码， 就大家互相知道了， 这个集群里面都有哪些成员，不同的成员的类型是什么， 是ps还是worker。
然后创建一个server，在server中会将ps以及所有worker的ip端口准备好，在同一台电脑开三个不同的端口，分别代表ps,chief supervisors和worker。角色的名称用strjob_name表示。从 tf.train.Server这个的定义开始，就每个角色不一样了。 如果角色名字是ps的话， 程序就join到这里，作为参数更新的服务， 等待其他worker角色给它提交参数更新的数据。如果是worker角色，就执行后面的计算任务。以ps为例(先创建ps文件)：
'''
(1)为每个角色添加IP地址和端口，创建server
'''
'''定义IP和端口号'''
#指定服务器ip和port
strps_hosts = '127.0.0.1:1234'
#指定两个终端的ip和port
strworker_hosts = '127.0.0.1:1235,127.0.0.1:1236'
#定义角色名称
strjob_name = 'ps'
task_index = 0
#将字符串转为数组
ps_hosts = strps_hosts.split(',')
worker_hosts = strworker_hosts.split(',')
cluster_spec= tf.train.ClusterSpec({'ps':ps_hosts,'worker':worker_hosts})
#创建Server
server = tf.train.Server(
cluster_spec,
job_name = strjob_name,
task_index = task_index)
3.2 为ps角色添加等待函数
ps角色使用server.join()函数进行线程挂起，开始接受连接消息。
'''
(2) 为ps角色添加等待函数
'''
#ps角色处于监听状态，等待终端连接
if strjob_name == 'ps':
print('waiting....')
server.join()
3.3 创建网络结构
与正常的程序不同，在创建网络结构时，使用tf.device()函数将全部的节点都放在当前任务下。task:0对应worker1(可以理解为任务0对应着角色1)，task:1对应worker2。
在rf.device()函数中的任务是通过tf.train.replica_device_setter()来指定的。
在tf.train.replica_device_setter()中使用worker_device()来定义具体任务名称:使用cluster的配置来指定角色和对应的ip地址，从而实现整个任务下的图节点，
'''
(3) 创建网络结构
'''
#设定训练集数据长度
n_train = 100
#生成x数据，[-1,1]之间，均分成n_train个数据
train_x = np.linspace(-1,1,n_train).reshape(n_train,1)
#把x乘以2，在加入(0,0.3)的高斯正太分布
train_y = 2*train_x + np.random.normal(loc=0.0,scale=0.3,size=[n_train,1])
#绘制x,y波形
plt.figure()
plt.plot(train_x,train_y,'ro',label='y=2x') #o使用圆点标记一个点
plt.legend()
plt.show()
#创建网络结构时，通过tf.device()函数将全部的节点都放在当前任务下 task:0对应worker1 task:1对应worker2
with tf.device(tf.train.replica_device_setter(
worker_device = '/job:worker/task:{0}'.format(task_index),
cluster = cluster_spec)):
'''
前向反馈
'''
#创建占位符
input_x = tf.placeholder(dtype=tf.float32)
input_y = tf.placeholder(dtype=tf.float32)
#模型参数
w = tf.Variable(tf.truncated_normal(shape=[1],mean=0.0,stddev=1),name='w') #设置正太分布参数 初始化权重
b = tf.Variable(tf.truncated_normal(shape=[1],mean=0.0,stddev=1),name='b') #设置正太分布参数 初始化偏置
#创建一个global_step变量
global_step = tf.train.get_or_create_global_step()
#前向结构
pred = tf.multiply(w,input_x) + b
#将预测值以直方图形式显示，给直方图命名为'pred'
tf.summary.histogram('pred',pred)
'''
反向传播bp
'''
#定义代价函数 选取二次代价函数
cost = tf.reduce_mean(tf.square(input_y - pred))
#将损失以标量形式显示 该变量命名为loss_function
tf.summary.scalar('loss_function',cost)
#设置求解器 采用梯度下降法 学习了设置为0.001 并把global_step变量放到优化器中，这样每运行一次优化器，globle_step就会自动获得当前迭代的次数
train = tf.train.GradientDescentOptimizer(learning_rate=0.001).minimize(cost,global_step = global_step)
saver = tf.train.Saver(max_to_keep = 1)
#合并所有的summary
merged_summary_op = tf.summary.merge_all()
#初始化所有变量，因此变量需要放在其前面定义
init =tf.global_variables_initializer()
为了使载入检查点文件能够同步循环次数，这里添加了一个global_step变量，将其放到优化器中。这样每运行一次优化器，global_step就会自动加1.
3.4 创建Supervisor，管理session
'''
(4)创建Supervisor，管理session
'''
training_epochs = 2000
display_step = 20
sv = tf.train.Supervisor(is_chief = (task_index == 0), #0号worker为chief
logdir='./LinearRegression/super/', #检查点和summary文件保存的路径
init_op = init, #初始化所有变量
summary_op = None, #summary_op用于自动保存summary文件，设置为None，表示不自动保存
saver = saver, #将保存检查点的saver对象传入，supervisor会自动保存检查点文件。否则设置为None
global_step = global_step,
save_model_secs = 50 #保存检查点文件的时间间隔
)
说明：
- 在tf.train.Supervisor()函数中，is_cheif表明了是否为cheif supervisors角色，这里将task_index = 0的worker设置成chief supervisors。
- logdir：为检查点和summary日志文件的保存路径。不过这个似乎启动就会去这个logdir的目录去看有没有checkpoint的文件， 有的话就自动装载了，没有就用init_op指定的初始化参数。
- init_op：表示使用初始化变量的函数。
- summary_op:将保存summary的对象传入，就会自动保存summary文件。这里设置为None,表示不自动保存。
- saver：将保存检查点的saver对象传入，Supervisor就会自动保存检查点文件。如果不想自动保存，就设置为None。
- global_step：为迭代次数。
- save_model_op：为保存检查点文件的时间间隔，这里设置成50，表明每50秒自动保存一次检查点文件。为了使程序运行时间长一些，我们更改了training_epochs参数。
3.5 迭代训练
session中的内容和之前的一样，直接迭代训练即可，由于使用了Supervisor管理session，将使用sv.summary_computed函数来保存summary文件，同样，如果想要手动保存监测点文件，也可以使用sv.saver.save()函数。
'''
(5) 迭代训练
'''
#连接目标角色创建session
with sv.managed_session(server.target) as sess:
print("sess ok：")
print(global_step.eval(session=sess))
print('开始迭代：')
#存放批次值和代价值
plotdata = {'batch_size':[],'loss':[]}
#开始迭代 这里step表示当前执行步数，迭代training_epochs轮 需要执行training_epochs*n_train步
for step in range(training_epochs*n_train):
for (x,y) in zip(train_x,train_y):
#开始执行图 并返回当前步数
_,step = sess.run([train,global_step],feed_dict={input_x:x,input_y:y})
#生成summary
summary_str = sess.run(merged_summary_op,feed_dict={input_x:x,input_y:y})
#将summary写入文件 手动保存summary日志文件
sv.summary_computed(sess,summary_str,global_step = step)
#一轮训练完成后 打印输出信息
if step % display_step == 0:
#计算代价值
loss = sess.run(cost,feed_dict={input_x:train_x,input_y:train_y})
print('step {0} cost {1} w {2} b{3}'.format(step,loss,sess.run(w),sess.run(b)))
#保存每display_step轮训练后的代价值以及当前迭代轮数
if not loss == np.nan:
plotdata['batch_size'].append(step)
plotdata['loss'].append(loss)
print('Finished!')
#手动保存检查点文件
#sv.saver.save(sess,'./LinearRegression/sv/sv.cpkt',global_step = step)
sv.stop()
在设置了自动保存检查点文件后，手动保存仍然有效。程序里我们在Supervisor对象创建的时候指定了自动保存检查点文件，程序里被我注释掉的最后一行是采用手动保存检查点文件。
在Supervisor对象创建的时候指定了不自动保存summary日志文件，我们采用了手动保存，调用了sv.summary_computed()函数。
在运行一半后终止，再运行Supervisor时会自动载入模型的参数，不需要手动调用saver.restore()。 在session中，不需要再运行tf.global_variables_initializer()函数。因为在Supervisor建立的时候回调用传入的init_op进行初始化，如果加了sess.run(tf.global_variables_initializer())，则会导致所载入模型的变量被二次清空。
3.6 建立worker文件
将ps.py文件复制两份，一个叫worker1.py,一个叫worker2.py。将角色名称修改为worker，并将worker2.py中的task_index修改为1。同时需要将worker2.py文件中手动保存summary日志的代码注释掉。
worker1.py文件修改如下：
#定义角色名称
strjob_name = 'worker'
task_index = 0
worker2.py文件修改如下：
#定义角色名称
strjob_name = 'worker'
task_index = 1
在这个程序中使用了sv.summary_computed()函数手动将运行时动态的数据保存下来，以便于在TensorBoard中查看，但是在分布式部署的时候，使用该功能还需要注意以下几点：
worker2文件中不能使用sv.summary_computed()函数，因为worker2不是chief supervisors，在worker2中是不会为Supervisor对象构造默认summary_writer(所有的summary日志信息都要通过该对象进行写)对象的，所以即使程序调用sv.summary_computed()也无法执行下去，程序会报错。 手写控制summary日志和检查点文件保存时，需要将chief supervisors以外的worker全部去掉才可以，可以使用Supervisor按时间间隔保存的形式来管理，这样用一套代码就可以解决了。
四、讨论
4.1. strworker_hosts
strworker_hosts = ‘127.0.0.1:1235,127.0.0.1:1236’
127.0.0.1为本地IP地址，我是给指定到了服务器具体的ip地址，然后分别在不同的节点上运行ps.py，worker1.py,worker2.py
4.2 ps可以不用读取数据文件
ps作为通信的节点，没必要跟worker节点一样，去读取数据文件。特别是我现在数据量大的时候，每个节点都读输入的数据文件，每个节点内存都要消耗100G左右，太恐怖了。
4.3 模型的调用
见下一篇笔记
4.4 如何与SGE集群配合着用
目前是分别运行ps.py, worker1.py,worker2.py ,每个脚本通过qsub的来提交，提交的时候指定好所用的集群和节点
五、代码详情
ps.py完整代码：
# -*- coding: utf-8 -*-
"""
Created on Thu Apr 19 08:52:30 2018
@author: zy
"""
import tensorflow as tf
import numpy as np
import os
import matplotlib.pyplot as plt
'''
分布式计算
'''
'''
(1)为每个角色添加IP地址和端口，创建server
'''
'''定义IP和端口号'''
#指定服务器ip和port
strps_hosts = '127.0.0.1:1234'
#指定两个终端的ip和port
strworker_hosts = '127.0.0.1:1235,127.0.0.1:1236'
#定义角色名称
strjob_name = 'ps'
task_index = 0
#将字符串转为数组
ps_hosts = strps_hosts.split(',')
worker_hosts = strworker_hosts.split(',')
cluster_spec = tf.train.ClusterSpec({'ps': ps_hosts,'worker': worker_hosts})
#创建server
server = tf.train.Server(
cluster_spec,
job_name = strjob_name,
task_index = task_index)
'''
(2) 为ps角色添加等待函数
'''
#ps角色处于监听状态，等待终端连接
if strjob_name == 'ps':
print('waiting....')
server.join()
'''
(3) 创建网络结构
'''
#设定训练集数据长度
n_train = 100
#生成x数据，[-1,1]之间，均分成n_train个数据
train_x = np.linspace(-1,1,n_train).reshape(n_train,1)
#把x乘以2，在加入(0,0.3)的高斯正太分布
train_y = 2*train_x + np.random.normal(loc=0.0,scale=0.3,size=[n_train,1])
#绘制x,y波形
plt.figure()
plt.plot(train_x,train_y,'ro',label='y=2x') #o使用圆点标记一个点
plt.legend()
plt.show()
#创建网络结构时，通过tf.device()函数将全部的节点都放在当前任务下
with tf.device(tf.train.replica_device_setter(
worker_device = '/job:worker/task:{0}'.format(task_index),
cluster = cluster_spec)):
'''
前向反馈
'''
#创建占位符
input_x = tf.placeholder(dtype=tf.float32)
input_y = tf.placeholder(dtype=tf.float32)
#模型参数
w = tf.Variable(tf.truncated_normal(shape=[1],mean=0.0,stddev=1),name='w') #设置正太分布参数 初始化权重
b = tf.Variable(tf.truncated_normal(shape=[1],mean=0.0,stddev=1),name='b') #设置正太分布参数 初始化偏置
#创建一个global_step变量
global_step = tf.train.get_or_create_global_step()
#前向结构
pred = tf.multiply(w,input_x) + b
#将预测值以直方图形式显示，给直方图命名为'pred'
tf.summary.histogram('pred',pred)
'''
反向传播bp
'''
#定义代价函数 选取二次代价函数
cost = tf.reduce_mean(tf.square(input_y - pred))
#将损失以标量形式显示 该变量命名为loss_function
tf.summary.scalar('loss_function',cost)
#设置求解器 采用梯度下降法 学习了设置为0.001 并把global_step变量放到优化器中，这样每运行一次优化器，global_step就会自动获得当前迭代的次数
train = tf.train.GradientDescentOptimizer(learning_rate=0.001).minimize(cost,global_step = global_step)
saver = tf.train.Saver(max_to_keep = 1)
#合并所有的summary
merged_summary_op = tf.summary.merge_all()
#初始化所有变量，因此变量需要放在其前面定义
init =tf.global_variables_initializer()
'''
(4)创建Supervisor，管理session
'''
training_epochs = 2000
display_step = 20
sv = tf.train.Supervisor(is_chief = (task_index == 0), #0号worker为chief
logdir='./LinearRegression/super/', #检查点和summary文件保存的路径
init_op = init, #初始化所有变量
summary_op = None, #summary_op用于自动保存summary文件，设置为None，表示不自动保存
saver = saver, #将保存检查点的saver对象传入，supervisor会自动保存检查点文件。否则设置为None
global_step = global_step,
save_model_secs = 50 #保存检查点文件的时间间隔
)
'''
(5) 迭代训练
'''
#连接目标角色创建session
with sv.managed_session(server.target) as sess:
print("sess ok：")
print(global_step.eval(session=sess))
print('开始迭代：')
#存放批次值和代价值
plotdata = {'batch_size':[],'loss':[]}
#开始迭代 这里step表示当前执行步数，迭代training_epochs轮 需要执行training_epochs*n_train步
for step in range(training_epochs*n_train):
for (x,y) in zip(train_x,train_y):
#开始执行图 并返回当前步数
_,step = sess.run([train,global_step],feed_dict={input_x:x,input_y:y})
#生成summary
summary_str = sess.run(merged_summary_op,feed_dict={input_x:x,input_y:y})
#将summary写入文件 手动保存summary日志文件
sv.summary_computed(sess,summary_str,global_step = step)
#一轮训练完成后 打印输出信息
if step % display_step == 0:
#计算代价值
loss = sess.run(cost,feed_dict={input_x:train_x,input_y:train_y})
print('step {0} cost {1} w {2} b{3}'.format(step,loss,sess.run(w),sess.run(b)))
#保存每display_step轮训练后的代价值以及当前迭代轮数
if not loss == np.nan:
plotdata['batch_size'].append(step)
plotdata['loss'].append(loss)
print('Finished!')
#手动保存检查点文件
#sv.saver.save(sess,'./LinearRegression/sv/sv.cpkt',global_step = step)
sv.stop()
worker1.py完整代码
# -*- coding: utf-8 -*-
"""
Created on Thu Apr 19 08:52:30 2018
@author: zy
"""
import tensorflow as tf
import numpy as np
import os
import matplotlib.pyplot as plt
'''
分布式计算
'''
'''
(1)为每个角色添加IP地址和端口，创建worker
'''
'''定义IP和端口号'''
#指定服务器ip和port
strps_hosts = '127.0.0.1:1234'
#指定两个终端的ip和port
strworker_hosts = '127.0.0.1:1235,127.0.0.1:1236'
#定义角色名称
strjob_name = 'worker'
task_index = 0
#将字符串转为数组
ps_hosts = strps_hosts.split(',')
worker_hosts = strworker_hosts.split(',')
cluster_spec = tf.train.ClusterSpec({'ps': ps_hosts,'worker': worker_hosts})
#创建server
server = tf.train.Server(
cluster_spec,
job_name = strjob_name,
task_index = task_index)
'''
(2) 为ps角色添加等待函数
'''
#ps角色处于监听状态，等待终端连接
if strjob_name == 'ps':
print('waiting....')
server.join()
'''
(3) 创建网络结构
'''
#设定训练集数据长度
n_train = 100
#生成x数据，[-1,1]之间，均分成n_train个数据
train_x = np.linspace(-1,1,n_train).reshape(n_train,1)
#把x乘以2，在加入(0,0.3)的高斯正太分布
train_y = 2*train_x + np.random.normal(loc=0.0,scale=0.3,size=[n_train,1])
#绘制x,y波形
plt.figure()
plt.plot(train_x,train_y,'ro',label='y=2x') #o使用圆点标记一个点
plt.legend()
plt.show()
#创建网络结构时，通过tf.device()函数将全部的节点都放在当前任务下
with tf.device(tf.train.replica_device_setter(
worker_device = '/job:worker/task:{0}'.format(task_index),
cluster = cluster_spec)):
'''
前向反馈
'''
#创建占位符
input_x = tf.placeholder(dtype=tf.float32)
input_y = tf.placeholder(dtype=tf.float32)
#模型参数
w = tf.Variable(tf.truncated_normal(shape=[1],mean=0.0,stddev=1),name='w') #设置正太分布参数 初始化权重
b = tf.Variable(tf.truncated_normal(shape=[1],mean=0.0,stddev=1),name='b') #设置正太分布参数 初始化偏置
#创建一个global_step变量
global_step = tf.train.get_or_create_global_step()
#前向结构
pred = tf.multiply(w,input_x) + b
#将预测值以直方图形式显示，给直方图命名为'pred'
tf.summary.histogram('pred',pred)
'''
反向传播bp
'''
#定义代价函数 选取二次代价函数
cost = tf.reduce_mean(tf.square(input_y - pred))
#将损失以标量形式显示 该变量命名为loss_function
tf.summary.scalar('loss_function',cost)
#设置求解器 采用梯度下降法 学习了设置为0.001 并把global_step变量放到优化器中，这样每运行一次优化器，global_step就会自动获得当前迭代的次数
train = tf.train.GradientDescentOptimizer(learning_rate=0.001).minimize(cost,global_step = global_step)
saver = tf.train.Saver(max_to_keep = 1)
#合并所有的summary
merged_summary_op = tf.summary.merge_all()
#初始化所有变量，因此变量需要放在其前面定义
init =tf.global_variables_initializer()
'''
(4)创建Supervisor，管理session
'''
training_epochs = 2000
display_step = 20
sv = tf.train.Supervisor(is_chief = (task_index == 0), #0号worker为chief
logdir='./LinearRegression/super/', #检查点和summary文件保存的路径
init_op = init, #初始化所有变量
summary_op = None, #summary_op用于自动保存summary文件，设置为None，表示不自动保存
saver = saver, #将保存检查点的saver对象传入，supervisor会自动保存检查点文件。否则设置为None
global_step = global_step,
save_model_secs = 50 #保存检查点文件的时间间隔
)
'''
(5) 迭代训练
'''
#连接目标角色创建session
with sv.managed_session(server.target) as sess:
print("sess ok：")
print(global_step.eval(session=sess))
print('开始迭代：')
#存放批次值和代价值
plotdata = {'batch_size':[],'loss':[]}
#开始迭代 这里step表示当前执行步数，迭代training_epochs轮 需要执行training_epochs*n_train步
for step in range(training_epochs*n_train):
for (x,y) in zip(train_x,train_y):
#开始执行图 并返回当前步数
_,step = sess.run([train,global_step],feed_dict={input_x:x,input_y:y})
#生成summary
summary_str = sess.run(merged_summary_op,feed_dict={input_x:x,input_y:y})
#将summary写入文件 手动保存summary日志文件
sv.summary_computed(sess,summary_str,global_step = step)
#一轮训练完成后 打印输出信息
if step % display_step == 0:
#计算代价值
loss = sess.run(cost,feed_dict={input_x:train_x,input_y:train_y})
print('step {0} cost {1} w {2} b{3}'.format(step,loss,sess.run(w),sess.run(b)))
#保存每display_step轮训练后的代价值以及当前迭代轮数
if not loss == np.nan:
plotdata['batch_size'].append(step)
plotdata['loss'].append(loss)
print('Finished!')
#手动保存检查点文件
#sv.saver.save(sess,'./LinearRegression/sv/sv.cpkt',global_step = step)
sv.stop()
worker2.py完整代码：
# -*- coding: utf-8 -*-
"""
Created on Thu Apr 19 08:52:30 2018
@author: zy
"""
import tensorflow as tf
import numpy as np
import os
import matplotlib.pyplot as plt
'''
分布式计算
'''
'''
(1)为每个角色添加IP地址和端口，创建worker
'''
'''定义IP和端口号'''
#指定服务器ip和port
strps_hosts = '127.0.0.1:1234'
#指定两个终端的ip和port
strworker_hosts = '127.0.0.1:1235,127.0.0.1:1236'
#定义角色名称
strjob_name = 'worker'
task_index = 1
#将字符串转为数组
ps_hosts = strps_hosts.split(',')
worker_hosts = strworker_hosts.split(',')
cluster_spec = tf.train.ClusterSpec({'ps': ps_hosts,'worker': worker_hosts})
#创建server
server = tf.train.Server(
cluster_spec,
job_name = strjob_name,
task_index = task_index)
'''
(2) 为ps角色添加等待函数
'''
#ps角色处于监听状态，等待终端连接
if strjob_name == 'ps':
print('waiting....')
server.join()
'''
(3) 创建网络结构
'''
#设定训练集数据长度
n_train = 100
#生成x数据，[-1,1]之间，均分成n_train个数据
train_x = np.linspace(-1,1,n_train).reshape(n_train,1)
#把x乘以2，在加入(0,0.3)的高斯正太分布
train_y = 2*train_x + np.random.normal(loc=0.0,scale=0.3,size=[n_train,1])
#绘制x,y波形
plt.figure()
plt.plot(train_x,train_y,'ro',label='y=2x') #o使用圆点标记一个点
plt.legend()
plt.show()
#创建网络结构时，通过tf.device()函数将全部的节点都放在当前任务下
with tf.device(tf.train.replica_device_setter(
worker_device = '/job:worker/task:{0}'.format(task_index),
cluster = cluster_spec)):
'''
前向反馈
'''
#创建占位符
input_x = tf.placeholder(dtype=tf.float32)
input_y = tf.placeholder(dtype=tf.float32)
#模型参数
w = tf.Variable(tf.truncated_normal(shape=[1],mean=0.0,stddev=1),name='w') #设置正太分布参数 初始化权重
b = tf.Variable(tf.truncated_normal(shape=[1],mean=0.0,stddev=1),name='b') #设置正太分布参数 初始化偏置
#创建一个global_step变量
global_step = tf.train.get_or_create_global_step()
#前向结构
pred = tf.multiply(w,input_x) + b
#将预测值以直方图形式显示，给直方图命名为'pred'
tf.summary.histogram('pred',pred)
'''
反向传播bp
'''
#定义代价函数 选取二次代价函数
cost = tf.reduce_mean(tf.square(input_y - pred))
#将损失以标量形式显示 该变量命名为loss_function
tf.summary.scalar('loss_function',cost)
#设置求解器 采用梯度下降法 学习了设置为0.001 并把global_step变量放到优化器中，这样每运行一次优化器，global_step就会自动获得当前迭代的次数
train = tf.train.GradientDescentOptimizer(learning_rate=0.001).minimize(cost,global_step = global_step)
saver = tf.train.Saver(max_to_keep = 1)
#合并所有的summary
merged_summary_op = tf.summary.merge_all()
#初始化所有变量，因此变量需要放在其前面定义
init =tf.global_variables_initializer()
'''
(4)创建Supervisor，管理session
'''
training_epochs = 2000
display_step = 20
sv = tf.train.Supervisor(is_chief = (task_index == 0), #0号worker为chief
logdir='./LinearRegression/super/', #检查点和summary文件保存的路径
init_op = init, #初始化所有变量
summary_op = None, #summary_op用于自动保存summary文件，设置为None，表示不自动保存
saver = saver, #将保存检查点的saver对象传入，supervisor会自动保存检查点文件。否则设置为None
global_step = global_step,
save_model_secs = 50 #保存检查点文件的时间间隔
)
'''
(5) 迭代训练
'''
#连接目标角色创建session
with sv.managed_session(server.target) as sess:
print("sess ok：")
print(global_step.eval(session=sess))
print('开始迭代：')
#存放批次值和代价值
plotdata = {'batch_size':[],'loss':[]}
#开始迭代 这里step表示当前执行步数，迭代training_epochs轮 需要执行training_epochs*n_train步
for step in range(training_epochs*n_train):
for (x,y) in zip(train_x,train_y):
#开始执行图 并返回当前步数
_,step = sess.run([train,global_step],feed_dict={input_x:x,input_y:y})
#生成summary
summary_str = sess.run(merged_summary_op,feed_dict={input_x:x,input_y:y})
#将summary写入文件 手动保存summary日志文件
#sv.summary_computed(sess,summary_str,global_step = step)
#一轮训练完成后 打印输出信息
if step % display_step == 0:
#计算代价值
loss = sess.run(cost,feed_dict={input_x:train_x,input_y:train_y})
print('step {0} cost {1} w {2} b{3}'.format(step,loss,sess.run(w),sess.run(b)))
#保存每display_step轮训练后的代价值以及当前迭代轮数
if not loss == np.nan:
plotdata['batch_size'].append(step)
plotdata['loss'].append(loss)
print('Finished!')
#手动保存检查点文件
#sv.saver.save(sess,'./LinearRegression/sv/sv.cpkt',global_step = step)
sv.stop()
参考资料
这里是一个广告位，，感兴趣的都可以发邮件聊聊：tiehan@sina.cn
![]()
个人公众号，比较懒，很少更新，可以在上面提问题，如果回复不及时，可发邮件给我： tiehan@sina.cn
