腳本之家服務(wù)器常用軟件

快捷導(dǎo)航

利用Tensorflow的隊列多線程讀取數(shù)據(jù)方式

更新時間：2020年02月05日 10:32:08 作者：心態(tài)已炸_沉迷學(xué)習(xí)

今天小編就為大家分享一篇利用Tensorflow的隊列多線程讀取數(shù)據(jù)方式，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧

在tensorflow中，有三種方式輸入數(shù)據(jù)

1. 利用feed_dict送入numpy數(shù)組

2. 利用隊列從文件中直接讀取數(shù)據(jù)

3. 預(yù)加載數(shù)據(jù)

其中第一種方式很常用，在tensorflow的MNIST訓(xùn)練源碼中可以看到，通過feed_dict={}，可以將任意數(shù)據(jù)送入tensor中。

第二種方式相比于第一種，速度更快，可以利用多線程的優(yōu)勢把數(shù)據(jù)送入隊列，再以batch的方式出隊，并且在這個過程中可以很方便地對圖像進(jìn)行隨機(jī)裁剪、翻轉(zhuǎn)、改變對比度等預(yù)處理，同時可以選擇是否對數(shù)據(jù)隨機(jī)打亂，可以說是非常方便。該部分的源碼在tensorflow官方的CIFAR-10訓(xùn)練源碼中可以看到，但是對于剛學(xué)習(xí)tensorflow的人來說，比較難以理解，本篇博客就當(dāng)成我調(diào)試完成后寫的一篇總結(jié)，以防自己再忘記具體細(xì)節(jié)。

讀取CIFAR-10數(shù)據(jù)集

按照第一種方式的話，CIFAR-10的讀取只需要寫一段非常簡單的代碼即可將測試集與訓(xùn)練集中的圖像分別讀?。?/p>

path = 'E:\Dataset\cifar-10\cifar-10-batches-py'
# extract train examples
num_train_examples = 50000
x_train = np.empty((num_train_examples, 32, 32, 3), dtype='uint8')
y_train = np.empty((num_train_examples), dtype='uint8')
for i in range(1, 6): 
 fpath = os.path.join(path, 'data_batch_' + str(i)) 
 (x_train[(i - 1) * 10000: i * 10000, :, :, :], y_train[(i - 1) * 10000: i * 10000])   = load_and_decode(fpath)
# extract test examples
fpath = os.path.join(path, 'test_batch')
x_test, y_test = load_and_decode(fpath)
return x_train, y_train, x_test, np.array(y_test)

其中l(wèi)oad_and_decode函數(shù)只需要按照CIFAR-10官網(wǎng)給出的方式decode就行，最終返回的x_train是一個[50000, 32, 32, 3]的ndarray，但對于ndarray來說，進(jìn)行預(yù)處理就要麻煩很多，為了取mini-SGD的batch，還自己寫了一個類，通過調(diào)用train_set.next_batch()函數(shù)來取，總而言之就是什么都要自己動手，效率確實不高

但對于第二種方式，讀取起來就要麻煩很多，但使用起來，又快又方便

首先，把CIFAR-10的測試集文件讀取出來，生成文件名列表

path = 'E:\Dataset\cifar-10\cifar-10-batches-py'
filenames = [os.path.join(path, 'data_batch_%d' % i) for i in range(1, 6)]

有了列表以后，利用tf.train.string_input_producer函數(shù)生成一個讀取隊列

filename_queue = tf.train.string_input_producer(filenames)

接下來，我們調(diào)用read_cifar10函數(shù)，得到一幅一幅的圖像，該函數(shù)的代碼如下：

def read_cifar10(filename_queue):
 label_bytes = 1
 IMAGE_SIZE = 32
 CHANNELS = 3
 image_bytes = IMAGE_SIZE*IMAGE_SIZE*3
 record_bytes = label_bytes+image_bytes

 # define a reader
 reader = tf.FixedLengthRecordReader(record_bytes)
 key, value = reader.read(filename_queue)
 record_bytes = tf.decode_raw(value, tf.uint8)

 label = tf.strided_slice(record_bytes, [0], [label_bytes])
 depth_major = tf.reshape(tf.strided_slice(record_bytes, [label_bytes],  
            [label_bytes + image_bytes]),
        [CHANNELS, IMAGE_SIZE, IMAGE_SIZE])
 image = tf.transpose(depth_major, [1, 2, 0])
 return image, label

第9行，定義一個reader，來讀取固定長度的數(shù)據(jù)，這個固定長度是由CIFAR-10數(shù)據(jù)集圖片的存儲格式?jīng)Q定的，1byte的標(biāo)簽加上32 *32 *3長度的圖像，3代表RGB三通道，由于圖片的是按[channel, height, width]的格式存儲的，為了變?yōu)槌Ｓ玫腫height, width, channel]維度，需要在17行reshape一次圖像，最終我們提取出了一副完整的圖像與對應(yīng)的標(biāo)簽

對圖像進(jìn)行預(yù)處理

我們?nèi)〕龅膇mage與label均為tensor格式，因此預(yù)處理將變得非常簡單

 if not distortion:
  IMAGE_SIZE = 32
 else:
  IMAGE_SIZE = 24
  # 隨機(jī)裁剪為24*24大小
  distorted_image = tf.random_crop(tf.cast(image, tf.float32), [IMAGE_SIZE, IMAGE_SIZE, 3])
  # 隨機(jī)水平翻轉(zhuǎn)
  distorted_image = tf.image.random_flip_left_right(distorted_image)
  # 隨機(jī)調(diào)整亮度
  distorted_image = tf.image.random_brightness(distorted_image, max_delta=63)
  # 隨機(jī)調(diào)整對比度
  distorted_image = tf.image.random_contrast(distorted_image, lower=0.2, upper=1.8)
  # 對圖像進(jìn)行白化操作，即像素值轉(zhuǎn)為零均值單位方差
  float_image = tf.image.per_image_standardization(distorted_image)

distortion是定義的一個輸入布爾型變量，默認(rèn)為True，表示是否對圖像進(jìn)行處理

填充隊列與隨機(jī)打亂

調(diào)用tf.train.shuffle_batch或tf.train.batch函數(shù)，以tf.train.shuffle_batch為例，函數(shù)的定義如下：

def shuffle_batch(tensors, batch_size, capacity, min_after_dequeue,
     num_threads=1, seed=None, enqueue_many=False, shapes=None,
     allow_smaller_final_batch=False, shared_name=None, name=None):

tensors表示輸入的張量(tensor)，batch_size表示要輸出的batch的大小，capacity表示隊列的容量，即大小，min_after_dequeue表示出隊操作后隊列中的最小元素數(shù)量，這個值是要小于隊列的capacity的，通過調(diào)整min_after_dequeue與capacity兩個變量，可以改變數(shù)據(jù)被隨機(jī)打亂的程度，num_threads表示使用的線程數(shù)，只要取大于1的數(shù)，隊列的效率就會高很多。

通常情況下，我們只需要輸入以上幾個變量即可，在CIFAR-10_input.py中，谷歌給出的代碼是這樣寫的：

if shuffle:
 images, label_batch = tf.train.shuffle_batch([image, label], batch_size,         min_queue_examples+3*batch_size,
       min_queue_examples, num_preprocess_threads)
else:
 images, label_batch = tf.train.batch([image, label], batch_size,
           num_preprocess_threads, 
           min_queue_examples + 3 * batch_size)

min_queue_examples由以下方式得到：

min_fraction_of_examples_in_queue = 0.4
min_queue_examples = int(NUM_EXAMPLES_PER_EPOCH_FOR_TRAIN 
       *min_fraction_of_examples_in_queue)

當(dāng)然，這些值均可以自己隨意設(shè)置,

最終得到的images，labels（label_batch），即為shape=[128, 32, 32, 3]的tensor，其中128為默認(rèn)batch_size。

激活隊列與處理異常

得到了images和labels兩個tensor后，我們便可以把這兩個tensor送入graph中進(jìn)行運(yùn)算了

# input tensor
img_batch, label_batch = cifar10_input.tesnsor_shuffle_input(batch_size)

# build graph that computes the logits predictions from the inference model
logits, predicts = train.inference(img_batch, keep_prob)

# calculate loss
loss = train.loss(logits, label_batch)

定義sess=tf.Session()后，運(yùn)行sess.run()，然而你會發(fā)現(xiàn)并沒有輸出，程序直接掛起了，仿佛死掉了一樣

原因是這樣的，雖然我們在數(shù)據(jù)流圖中加入了隊列，但只有調(diào)用tf.train.start_queue_runners()函數(shù)后，數(shù)據(jù)才會動起來，被負(fù)責(zé)輸入管道的線程填入隊列，否則隊列將會掛起。

OK，我們調(diào)用函數(shù)，讓隊列運(yùn)行起來

with tf.Session(config=run_config) as sess:
 sess.run(init_op) # intialization
 queue_runner = tf.train.start_queue_runners(sess)
 for i in range(10):
  b1, b2 = sess.run([img_batch, label_batch])
  print(b1.shape)

在這里為了測試，我們?nèi)?0次輸出，看看輸出的batch1的維度是否正確

10個batch的維度均為正確的，但是tensorflow卻報了錯，錯誤的文字內(nèi)容如下：

2017-12-19 16:40:56.429687: W C:\tf_jenkins\home\workspace\rel-win\M\windows-gpu\PY\36\tensorflow\core\kernels\queue_base.cc:295] _ 0 _ input_producer: Skipping cancelled enqueue attempt with queue not closed

簡單地看一下，大致意思是說我們的隊列里還有數(shù)據(jù)，但是程序結(jié)束了，拋出了異常，因此，我們還需要定義一個Coordinator，也就是協(xié)調(diào)器來處理異常

Coordinator有3個主要方法：

1. tf.train.Coordinator.should_stop() 如果線程應(yīng)該停止，返回True

2. tf.train.Coordinator.request_stop() 請求停止線程

3. tf.train.Coordinator.join() 等待直到指定線程停止

首先，定義協(xié)調(diào)器

coord = tf.train.Coordinator()

將協(xié)調(diào)器應(yīng)用于QueueRunner

queue_runner = tf.train.start_queue_runners(sess, coord=coord)

結(jié)束數(shù)據(jù)的訓(xùn)練或測試后，關(guān)閉線程

coord.request_stop()
coord.join(queue_runner)

最終的sess代碼段如下：

coord = tf.train.Coordinator()
with tf.Session(config=run_config) as sess:
 sess.run(init_op)
 queue_runner = tf.train.start_queue_runners(sess, coord=coord)
 for i in range(10):
  b1, b2 = sess.run([img_batch, label_batch])
  print(b1.shape)
 coord.request_stop()
 coord.join(queue_runner)

得到的輸出結(jié)果為：