第八节，配置分布式TensorFlow

 因为随着神经网络层数的增多，须要训练的参数也会增多，随之而来须要的数据集就会很大，这样会形成须要更大的运算资源，并且还要消耗很长的运算时间。TensorFlow提供了一个能够分布式部署的模式，将一个训练任务拆分红多个小任务，配置到不一样的计算机上完成协同运算，这样使用计算机群运算来代替单机运算，可使训练时间大幅度缩短。

一 分布式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。代码主要分为如下几部分：

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)

2.为ps角色添加等待函数

ps角色使用server.join()函数进行线程挂起，开始接受链接消息。

'''
(2) 为ps角色添加等待函数
'''
#ps角色处于监听状态，等待终端链接
if strjob_name == 'ps':
    print('waiting....')
    server.join()   

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.

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参数。

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())，则会致使所载入模型的变量被二次清空。

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按时间间隔保存的形式来管理，这样用一套代码就能够解决了。

7.部署运行

在spyder中先将ps.py文件运行起来，选择菜单Consoles->Open an Ipython console，新打开一个Consoles，以下图

在spider面板右下角，能够看到在原有标签为'Console 1/A'标签又多了一个‘Console 2/A’标签，选中这个标签，就激活了这个标签。

运行worker2.py文件。同理，启动'Console 3/A'运行worker1.py文件。

下面咱们能够看到worker1.py文件的输出：

咱们在程序中设置display_step为20，即迭代20次输出一次信息，咱们可能看到这个输出并非连续的，这是由于跳过的步骤被分配到了worker2中去运算了。

worker2.py文件对应的窗口显示的信息以下：

从图中能够看到worker2和chief supervisors的迭代顺序是互补，但也有多是没有绝对互补的，可是为何有时候没有绝对互补？可能与Supervisor中的同步算法有关。

分布运算的目的是为了提升总体运算速度，若是同步epoch的准确度须要以牺牲整体运算速度为代价，天然很不合适。因此更合理的推断是由于单机单次运算太快迫使算法使用了更宽松的同步机制。

重要的一点是对于指定步数的学习参数w和b是一致的。即统一迭代论述的值是同样的，这代表两个终端是在相同的起点上进行运算的。

对于ps.py文件，其对应的窗口一直默默的只显示打印的那句话waiting....，由于它只负责链接参与运算。

四 最后再补充一些名词解释

客户端(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) 。
  

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

View Code

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

View Code

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

View Code