本博文參考TensorFlow實戰Google深度學習框架（鄭澤宇，顧思宇），僅用作學習

一、TFRecord輸入資料格式

TFRecord是tensorflow中儲存資料的統一格式。可以統一不同的原始資料格式，並更加有效地管理不同的屬性。TFRecord檔案中的資料都是通過tf.train.Example Protocol Buffer的格式儲存的。是一種可將影象的資料和標籤放在一起的二進位制檔案，能節省記憶體，在TensorFlow中快速讀取儲存。

tf.train.Example的定義如下：

message Example{
    Features Features=1;
};

message Features{
    map<string,Feature> feature=1;
};

message Feature{
    oneof kind{
       ByteList bytes_list=1;
       FloatList float_list=2;
       Int64List int64_list=3;
    }
};
 

tf.train.Example中包含了一個從屬性名稱到取值的字典。屬性名稱為字串，屬性的取值可以為字串（ByteList），實數列表（FloatList）或整數列表（Int64List）。

從檔案中讀取資料一般分為：把樣本資料寫入TFRecords二進位制檔案，再從佇列中讀取。

1、生成TFRecord檔案

需要將資料填到tf.train.Example的協議快取區（Protocol Buffer）中，將協議快取區序列化為一個字串，通過tf.python_io.TFRecordWriter寫入TFRecord檔案中。

import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import numpy as np

mnist=input_data.read_data_sets("/path/to/mnist/data",dtype=tf.unit8,one_hot=True)
#訓練資料
image=mnist.train.images
#訓練資料所對應的的正確答案，可以作為一個屬性儲存在TFRecorde中
labels=mnist.train.labels
#訓練資料的影象解析度，可以作為Example中的一個屬性
pixels=image.shape[1]
num_examples=mnist.train.num_examples

#生成整數型的屬性
def _int64_feature(value):
    return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))
#生成字串型的屬性
def _bytes_feature(value):
    return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))

filename="/path/to/outpput.tfrecords"
writer=tf.python_io.TFRecordWriter(filename)
for i in range(num_examples):
    #將影象矩陣轉化為一個字串
    image_raw=image[i].tostring()
    #將一個樣例轉化為ExamplecProtocol Buffer，並將所有資訊寫入這個資料結構
    example=tf.train.Example(features=tf.train.Features(feature={
        'pixels': _int64_feature(pixels),
        'labels': _int64_feature(np.argmax(labels[i])),
        'image_raw':_bytes_feature(image_raw)
    }))
 

    writer.write(example.SerializeToString())
writer.close()

首先建立張量，從二進位制檔案中讀取一個樣本

建立張量，從二進位制檔案中隨機讀取一個mini-batch

把每一批張量傳入網路作為輸入節點

import tensorflow as tf

# 讀取檔案。
files = tf.train.match_filenames_once("/path/to/output.tfrecords")
filename_queue = tf.train.string_input_producer(files, shuffle=False)
reader = tf.TFRecordReader()
#從檔案中讀出一個樣例（也可以永別的函式讀取多個樣例）
_,serialized_example = reader.read(filename_queue)

# 解析讀取的樣例（也可以用別的函式解析多個樣例）
features = tf.parse_single_example(
      serialized_example,
      features={
        'image_raw':FixedLenFeature([],tf.string),
        'labels': FixedLenFeature([],tf.int64),
        'pixels': FixedLenFeature([],tf.int64)
      })
 


images=tf.decode_raw(features['image_raw'],tf.uint8)
labels=tf.cast(features['labels'],tf.int32)
pixels=tf.cast(features['pixels'],tf.int32)

sess=tf.Session()
#啟動多執行緒處理資料
coord=tf.train.Coordinator()
threads=tf.train.start_queue_runners(sess=sess,coord=coord)

for i in range(10):
    image,label,pixel=sess.run([images,labels,pixels])

二、佇列

佇列也是圖中的一個節點

佇列主要有FIFOQueue和RandomShuffleQueue。tensorflow可以利用佇列來實現多執行緒輸入資料處理。

FIFOQueue建立一個先入先出佇列。RandomShuffleQueue建立一個隨機佇列，在非同步訓練中很重要

API：tf.FIFOQueue   tf.RandomShuffleQueue

import tensorflow as tf
#建立先進先出佇列，指定佇列最多可以儲存兩個元素
q=tf.FIFOQueue(2,"int32")
#初始化佇列的元素
init=q.enqueue_many(([0,10],))
#將佇列的第一個元素出列，儲存在x裡面
x=q.dequeue()
y=x+1
#將y加入佇列
q_inc=q.enqueue([y])

with tf.Session() as sess:
    #執行初始化佇列的操作
    init.run()
    #執行出佇列，將佇列的元素加1，加1後的元素進隊的迴圈操作5次
    for _ in range(5):
       v,_=sess.run([x,q_inc])
       print(v)

三、佇列管理器&執行緒和協調器

tensorflow提供了tf.Cooradinator和tf.QueueRunner兩個類來完成多執行緒協同的功能。

QueueRunner：佇列管理器

coordinator：協調器，協調執行緒間的關係可以視為一種訊號量，用來做同步。主要用於協同多個執行緒一起停止，並提供了should_stop、request_stop和join三個函式

啟動執行緒之前，需要先宣告一個tf.Cooradinator類，並將這個類傳入每一個建立的執行緒中。

啟動的執行緒需要一直查詢tf.Cooradinator類中提供的should_stop函式，返回值為True時，執行緒也退出。

每個執行緒都可以通過request_stop函式來通知其他執行緒退出。當執行緒呼叫request_stop時，should_stop的返回值將被設定為True，這樣其他的執行緒就可以同時終止了。

import tensorflow as tf
import numpy as np
import threading
import time

#執行緒中執行的程式，每隔一秒判斷是否需要停止並列印自己的ID
def MyLoop(coord,worker_id):
    #使用tf.Coordinator類提供的協同工具判斷當前執行緒是否需要停止
    while not coord.should_stop():
        #隨機停止所有的執行緒
        if np.random.rand()<0.1:
            print("stoping from id: %d\n" % worker_id)
            #呼叫coord.request_stop()函式來通知其他執行緒停止
            coord.request_stop()
        else:
            #列印當前執行緒ID
            print("working on id: %d\n" % worker_id)
        #暫停1秒
        time.sleep(1)
#宣告一個協調器來協同多個執行緒
coord=tf.train.Coordinator()

#建立5個執行緒
threads=[
    threading.Thread(target=MyLoop,args=(coord,i,)) for  i in range(5)]
#啟動所有執行緒
for t in threads:
    t.start()
#join操作等待所有執行緒關閉，這一函式才能返回
coord.join(threads)

四、輸入資料處理框架

import tensorflow as tf
files=tf.train.match_filenames_once("/path/to/output.tfrecords")
filename_queue = tf.train.string_input_producer(files, shuffle=False)
reader = tf.TFRecordReader()
_, serialized_example = reader.read(filename_queue)
features = tf.parse_single_example(
      serialized_example,
      features={
          'image': tf.FixedLenFeature([], tf.string),
          'label': tf.FixedLenFeature([], tf.int64),
          'height': tf.FixedLenFeature([], tf.int64),
          'width': tf.FixedLenFeature([], tf.int64),
          'channels': tf.FixedLenFeature([], tf.int64),
      })
image,label=features['image'],features['label']
height,width=features['height'],features['width']
channels=features['channels']

decoded_image=tf.decode_raw(image,tf.uint8)
decoded_image.set_shape([height,width,channels])

image_size=299
distorted_image=preprocess_for_train(decoded_image,image_size,image_size,None)

min_after_dequeue=10000
batch_size=100
capacity=min_after_dequeue+3*batch_size
image_batch,label_batch=tf.train.shuffle_batch(
    [distorted_image,label],batch_size=batch_size,
    capacity=capacity,min_after_dequeue=min_after_dequeue)

logit=inference(image_batch)
loss=calc_loss(logit,label_batch)
train_step=tf.train.GradientDescentOptimizer(learning_rate)\
    .minimize(loss)

with tf.Session() as sess:
        tf.global_variables_initializer().run()
        coord = tf.train.Coordinator()
        threads = tf.train.start_queue_runners(sess=sess, coord=coord)
        for i in range(TRAINING_ROUDNS):
            sess.run(train_step)
        coord.request_stop()
        coord.join(threads)