python - tf.train.shuffle_batch 和 `tf.train.batch 发生了什么？

Question

我使用 Binary data训练 DNN。

但是 tf.train.shuffle_batch 和 tf.train.batch 让我很困惑。

这是我的代码，我将对其进行一些测试。

首先Using_Queues_Lib.py:

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import os

from six.moves import xrange  # pylint: disable=redefined-builtin
import tensorflow as tf

NUM_EXAMPLES_PER_EPOCH_FOR_TRAIN = 100
REAL32_BYTES=4


def read_dataset(filename_queue,data_length,label_length):
  class Record(object):
    pass
  result = Record()

  result_data  = data_length*REAL32_BYTES
  result_label = label_length*REAL32_BYTES
  record_bytes = result_data + result_label

  reader = tf.FixedLengthRecordReader(record_bytes=record_bytes)
  result.key, value = reader.read(filename_queue)

  record_bytes = tf.decode_raw(value, tf.float32)
  result.data  = tf.strided_slice(record_bytes, [0],[data_length])#record_bytes: tf.float list
  result.label = tf.strided_slice(record_bytes, [data_length],[data_length+label_length])
  return result


def _generate_data_and_label_batch(data, label, min_queue_examples,batch_size, shuffle):
  num_preprocess_threads = 16   #only speed code
  if shuffle:
    data_batch, label_batch = tf.train.shuffle_batch([data, label],batch_size=batch_size,num_threads=num_preprocess_threads,capacity=min_queue_examples + batch_size,min_after_dequeue=min_queue_examples)
  else:
    data_batch, label_batch = tf.train.batch([data, label],batch_size=batch_size,num_threads=num_preprocess_threads,capacity=min_queue_examples + batch_size)
  return data_batch, label_batch

def inputs(data_dir, batch_size,data_length,label_length):
  filenames = [os.path.join(data_dir, 'test_data_SE.dat')]
  for f in filenames:
    if not tf.gfile.Exists(f):
      raise ValueError('Failed to find file: ' + f)

  filename_queue = tf.train.string_input_producer(filenames)

  read_input = read_dataset(filename_queue,data_length,label_length)

  read_input.data.set_shape([data_length])   #important
  read_input.label.set_shape([label_length]) #important


  min_fraction_of_examples_in_queue = 0.4
  min_queue_examples = int(NUM_EXAMPLES_PER_EPOCH_FOR_TRAIN *
                       min_fraction_of_examples_in_queue)
  print ('Filling queue with %d samples before starting to train. '
     'This will take a few minutes.' % min_queue_examples)

  return _generate_data_and_label_batch(read_input.data, read_input.label,
                                     min_queue_examples, batch_size,
                                     shuffle=True)

第二个Using_Queues.py:

import Using_Queues_Lib
import tensorflow as tf
import numpy as np
import time


max_steps=10
batch_size=100
data_dir=r'.'
data_length=2
label_length=1

#-----------Save paras-----------
import struct
def WriteArrayFloat(file,data):
  fout=open(file,'wb')        
  fout.write(struct.pack('<'+str(data.flatten().size)+'f',
                                *data.flatten().tolist()))
  fout.close()
#-----------------------------

def add_layer(inputs, in_size, out_size, activation_function=None):
  Weights = tf.Variable(tf.truncated_normal([in_size, out_size]))
  biases = tf.Variable(tf.zeros([1, out_size]) + 0.1)
  Wx_plus_b = tf.matmul(inputs, Weights) + biases
  if activation_function is None:
    outputs = Wx_plus_b
  else:
    outputs = activation_function(Wx_plus_b)
  return outputs

data_train,labels_train=Using_Queues_Lib.inputs(data_dir=data_dir,
                      batch_size=batch_size,data_length=data_length,
                                          label_length=label_length)

xs=tf.placeholder(tf.float32,[None,data_length])
ys=tf.placeholder(tf.float32,[None,label_length])

l1 = add_layer(xs, data_length, 5, activation_function=tf.nn.sigmoid)
l2 = add_layer(l1, 5, 5, activation_function=tf.nn.sigmoid)
prediction = add_layer(l2, 5, label_length, activation_function=None)

loss = tf.reduce_mean(tf.square(ys - prediction))
train_step = tf.train.GradientDescentOptimizer(0.2).minimize(loss)

sess=tf.InteractiveSession()
tf.global_variables_initializer().run()

tf.train.start_queue_runners()

for i in range(max_steps):
  start_time=time.time()
  data_batch,label_batch=sess.run([data_train,labels_train])
  sess.run(train_step, feed_dict={xs: data_batch, ys: label_batch})
  duration=time.time()-start_time
  if i % 1 == 0:
    example_per_sec=batch_size/duration
    sec_pec_batch=float(duration)
    WriteArrayFloat(r'./data/'+str(i)+'.bin',
        np.concatenate((data_batch,label_batch),axis=1))
    format_str=('step %d,loss=%.8f(%.1f example/sec;%.3f sec/batch)')
    loss_value=sess.run(loss, feed_dict={xs: data_batch, ys: label_batch})
    print(format_str%(i,loss_value,example_per_sec,sec_pec_batch))

here中的数据. 它由 Mathematica 生成。

data = Flatten@Table[{x, y, x*y}, {x, -1, 1, .05}, {y, -1, 1, .05}];
BinaryWrite[file, mydata, "Real32", ByteOrdering -> -1];
Close[file];

数据长度:1681

数据如下所示:

绘制数据:红色 到绿色 颜色表示它们在 here 中出现的时间

运行 Using_Queues.py，它将产生十个批处理，我在这张图中绘制每个 bach:(batch_size=100 和 min_queue_examples=40)

如果 batch_size=1024 和 min_queue_examples=40:

如果 batch_size=100 和 min_queue_examples=4000:

如果 batch_size=1024 和 min_queue_examples=4000:

即使 batch_size=1681 和 min_queue_examples=4000:

该区域没有充满点。

为什么？

那么，为什么要改变 min_queue_examples 来增加随机性呢？如何确定min_queue_examples的值？

tf.train.shuffle_batch 发生了什么？

Answer 1

tf.train.shuffle_batch() 的采样函数(因此 tf.RandomShuffleQueue )使用有点微妙。实现使用 tf.RandomShuffleQueue.dequeue_many(batch_size) ，其(简化)实现如下:

• 当出队的元素数量小于batch_size时:
  • 等到队列至少包含 min_after_dequeue + 1 个元素。
  • 从队列中随机均匀选择一个元素，将其从队列中移除，加入输出批处理。

另一件需要注意的事情是如何将元素添加到队列中，它使用运行 tf.RandomShuffleQueue.enqueue() 的后台线程。在同一个队列上:

• 等到队列的当前大小小于它的容量。
• 将元素添加到队列中。

因此，队列的capacity 和min_after_dequeue 属性(加上入队的输入数据的分布)决定了tf. train.shuffle_batch() 将采样。输入文件中的数据似乎是有序的，因此您完全依赖于 tf.train.shuffle_batch() 函数来实现随机性。

依次进行可视化:

1. 如果 capacity 和 min_after_dequeue 相对于数据集较小，“洗牌”将从类似于“滑动窗口”的小群体中选择随机元素数据集。您会在出列的批处理中看到旧元素的概率很小。

2. 如果相对于数据集，batch_size 较大而 min_after_dequeue 较小，则“洗牌”将再次从跨数据集的小“滑动窗口”中进行选择数据集。

3. 如果 min_after_dequeue 相对于 batch_size 和数据集的大小而言较大，您将看到(近似)来自每批数据的均匀样本。

4. 如果 min_after_dequeue 和 batch_size 相对于数据集的大小而言较大，您将看到(近似)来自每批数据的均匀样本。

5. 在min_after_dequeue为4000，batch_size为1681的情况下，注意队列中每个元素采样时期望的副本数是4000/1681 = 2.38，因此更有可能对某些元素进行多次采样(并且不太可能对每个唯一元素仅采样一次)。