有没有办法在分布式 Tensorflow 中附加数据？答案

【问题标题】：Is there a way to append data in distributed Tensorflow?有没有办法在分布式 Tensorflow 中附加数据？
【发布时间】：2018-08-16 01:33:55
【问题描述】：

我使用分布式 TensorFlow 不是为了分发网络，而是为了分发工作。

通过分布式 TensorFlow，我们获得了分配工作的框架以及工作人员之间的状态通信。这种轻量级的加权通信协议、内置的恢复和特定任务的设备选择使我尝试使用分布式 tensorFlow 来并行构建多个微模型。

所以在我的代码中，这就是我正在做的事情。

def main(_): 
    #some global data block
    a = np.arange(10).reshape((5, 2))
    with tf.device(tf.train.replica_device_setter(worker_device="/job:worker/task:%d" %server.server_def.task_index,cluster=server.server_def.cluster)):
        #some ops to keep the cluster alive 
        var = tf.Variable(initial_value=10, name='var')
        op = tf.assign_add(var,10)
        xx = tf.placeholder("float")
        yy = tf.reduce_sum(xx)
        #start monitoring session 
    with tf.train.MonitoredTrainingSession(master=server.target,is_chief=is_chief) as mon_sess:
        mon_sess.run(op)
        #distribute data
        inputs = a[:,server.server_def.task_index]
        #start a local session in worker 
        sess = tf.Session()
        sum_value = sess.run(yy,feed_dict={xx:inputs})
        sess.close()

在每个工人的工作完成后，我想将一些信息附加到全局网络中的一个变量。（由于我们无法更新上面例子中a这样的全局变量，我想利用mon_sess来更新全局网络。

我想继续附加一些张量（每个工人的 o/p）并让chief 读取和写出它。有没有办法做到这一点？

如果您在上述方法中发现任何问题，请更新。

谢谢，

【问题讨论】：

标签： python tensorflow distributed

【解决方案1】：

我厌倦了这个并且能够从本地工人信息获取更新到全球网络

import tensorflow as tf
import numpy as np
import os
import time
def main(server, log_dir, context):
    #create a random array
    a = np.arange(10).reshape((5, 2))
    with tf.device(tf.train.replica_device_setter(worker_device="/job:worker/task:%d" %server.server_def.task_index,cluster=server.server_def.cluster)):
        var = tf.Variable(initial_value=10, name='var')
        op = tf.assign_add(var,10)
        xx = tf.placeholder("float")
        yy = tf.reduce_sum(xx)
        concat_init = tf.Variable([0],dtype=tf.float32)
        sum_holder = tf.placeholder(tf.float32)
        concat_op = tf.concat([concat_init,sum_holder],0)
        assign_op = tf.assign(concat_init,concat_op,validate_shape=False)
    is_chief = server.server_def.task_index == 0
    with tf.train.MonitoredTrainingSession(master=server.target,is_chief=is_chief) as mon_sess:
        mon_sess.run(op)
        print (a)
        print ("reading my part")
        inputs = a[:,server.server_def.task_index]
        print(inputs)
        sess = tf.Session()
        sum_value = sess.run(yy,feed_dict={xx:inputs})
        print(sum_value)
        mon_sess.run(assign_op,feed_dict={sum_holder:[sum_value]})
        if is_chief:
            time.sleep(5)
            worker_sums = mon_sess.run(assign_op,feed_dict={sum_holder:[0]})
            print (worker_sums)
        sess.close()
        if is_chief:
            while True:
                pass

【讨论】：