tensorflow數據輸入（文件輸入）

准備數據

前期已經將數據生成了tfrecord格式，這里主要是研究如果從tfrecord格式文件中讀取數據batch

讀文件名

獲取tf_record格式的文件名列表

tf_record_pattern = os.path.join(FLAGS.data_dir,'%s-*' %self.subset)  # subset in ['train','validation']
data_files = tf.gfile.Glob(tf_record_pattern)
或者
1 data_files = tf.train.match_filenames_once(tf.record_pattern)

文件名隊列

用tf.train.string_input_prodecer函數來生成一個隊列，放置文件名，之后文件閱讀器會需要它來讀取數據

注意：並不是一次性將所有文件名都放進隊列，它是有容量大小限制的，也就是說隊列大小為capacity

string_tensor:   文件名列表，一般就是類似 train-00000-of-00008這種

num_epochs=None:  就是說文件名列表放進隊列后是循環無限次，如果設定了特定的值,那么整個隊列只能循環num_epochs次，這里epoch就是指遍歷所有文件名一次，也相當於說對所有數據進行迭代一個epoch

shuffle:  如果為真的話，文件名在每個epoch中都會隨機重排

capacity:  隊列的大小

后面三個參數都不重要

Return: A queue with the output strings. A Queue is added to the current Graph's QUEUE_RUNNER collection.

這個QueueRunner的工作線程是獨立於文件閱讀器的線程，因此亂序和將文件名推入到文件名隊列這些過程不會阻塞文件閱讀器運行。

tf.train.slice_input_producer

這里,如果直接准備的是數據的話，不需要文件名隊列，可以直接產生數據隊列

看函數名也可以看出隊列中的數據時一條slice數據（打亂后shuffle=True），然后拿tf.train.batch()直接取batch就可以獲得亂序batch

 文件讀取器

根據不同的文件格式，來選擇對應的文件閱讀器，然后將文件名隊列提供給閱讀器的read方法。閱讀器的read方法會輸出一個key來表征輸入的文件和一個字符串標量解碼成張量從而構造成樣本。

1. 從csv文件中讀取數據，需要使用TextLineReader和decode_csv操作

filename_queue = tf.train.string_input_producer(["file0.csv", "file1.csv"])

reader = tf.TextLineReader()
key, value = reader.read(filename_queue)

# Default values, in case of empty columns. Also specifies the type of the
# decoded result.
record_defaults = [[1], [1], [1], [1], [1]]
col1, col2, col3, col4, col5 = tf.decode_csv(
    value, record_defaults=record_defaults)
features = tf.concat(0, [col1, col2, col3, col4])

with tf.Session() as sess:
  # Start populating the filename queue.
  coord = tf.train.Coordinator()
  threads = tf.train.start_queue_runners(coord=coord)

  for i in range(1200):
    # Retrieve a single instance:
    example, label = sess.run([features, col5])

  coord.request_stop()
  coord.join(threads)

每次read的執行都會從文件中讀取一行內容， decode_csv 操作會解析這一行內容並將其轉為張量列表。如果輸入的參數有缺失，record_default參數可以根據張量的類型來設置默認值。

在調用run或者eval去執行read之前， 你必須調用tf.train.start_queue_runners來將文件名填充到隊列。否則read操作會被阻塞到文件名隊列中有值為止。

2. 從二進制文件中讀取固定長度記錄，可以使用tf.FixedLengthRecordReader的tf.decode_raw操作。decode_raw操作可以講一個字符串轉換為一個uint8的張量。

舉例來說，the CIFAR-10 dataset的文件格式定義是：每條記錄的長度都是固定的，一個字節的標簽，后面是3072字節的圖像數據。uint8的張量的標准操作就可以從中獲取圖像片並且根據需要進行重組。 例子代碼可以在tensorflow/models/image/cifar10/cifar10_input.py找到

3.標准tensorflow格式，將數據轉換成TFRecords文件

從TFRecords文件中讀取數據， 可以使用tf.TFRecordReader的tf.parse_single_example解析器。這個parse_single_example操作可以將Example協議內存塊(protocol buffer)解析為張量

批處理

在數據輸入管線的末端， 我們需要有另一個隊列來執行輸入樣本的訓練，評價和推理。因此我們使用tf.train.shuffle_batch 函數來對隊列中的樣本進行亂序處理

def read_my_file_format(filename_queue):
  reader = tf.SomeReader()
  key, record_string = reader.read(filename_queue)
  example, label = tf.some_decoder(record_string)
  processed_example = some_processing(example)
  return processed_example, label

def input_pipeline(filenames, batch_size, num_epochs=None):
  filename_queue = tf.train.string_input_producer(
      filenames, num_epochs=num_epochs, shuffle=True)
  example, label = read_my_file_format(filename_queue)　　#這里是單個reader在讀文件，batch是當個reader讀單個文件shuffle成batch
  # min_after_dequeue defines how big a buffer we will randomly sample
  #   from -- bigger means better shuffling but slower start up and more
  #   memory used.
  # capacity must be larger than min_after_dequeue and the amount larger
  #   determines the maximum we will prefetch.  Recommendation:
  #   min_after_dequeue + (num_threads + a small safety margin) * batch_size
  min_after_dequeue = 10000
  capacity = min_after_dequeue + 3 * batch_size
  example_batch, label_batch = tf.train.shuffle_batch(
      [example, label], batch_size=batch_size, capacity=capacity,
      min_after_dequeue=min_after_dequeue)
  return example_batch, label_batch

tf.train.shuffle_batch()

從上面函數定義看出shuffle_batch的輸入是tensors，看代碼發現不是直接從queue中獲取batch，而是以一種自定義的方式讀取tensor后再獲取batch

如果你需要對不同文件中的樣子有更強的亂序和並行處理，可以使用tf.train.shuffle_batch_join 函數. 示例:

def read_my_file_format(filename_queue):
  # Same as above

def input_pipeline(filenames, batch_size, read_threads, num_epochs=None):
  filename_queue = tf.train.string_input_producer(
      filenames, num_epochs=num_epochs, shuffle=True) 
 7   example_list = [read_my_file_format(filename_queue)
                  for _ in range(read_threads)]           #這里是多個reader，batch_join將是多個reader讀多個文件然后構成batch
  min_after_dequeue = 10000
  capacity = min_after_dequeue + 3 * batch_size
  example_batch, label_batch = tf.train.shuffle_batch_join(
      example_list, batch_size=batch_size, capacity=capacity,
      min_after_dequeue=min_after_dequeue)
  return example_batch, label_batch

 在這個例子中， 雖然只使用了一個文件名隊列， 但是TensorFlow依然能保證多個文件閱讀器從同一次迭代(epoch)的不同文件中讀取數據，直到這次迭代的所有文件都被開始讀取為止。(通常來說一個線程來對文件名隊列進行填充的效率是足夠的) 

另一種替代方案是：使用tf.train.shuffle_batch 函數,設置num_threads的值大於1。 這種方案可以保證同一時刻只在一個文件中進行讀取操作(但是讀取速度依然優於單線程)，而不是之前的同時讀取多個文件。這種方案的優點是：

• 避免了兩個不同的線程從同一個文件中讀取同一個樣本。
• 避免了過多的磁盤搜索操作。

 創建線程並使用QueueRunner對象來預取

使用上面的tf.train.string_input_producer和tf.train.shuffle_batch_join等函數添加QueueRunner到你的數據流圖中。在你運行任何訓練步驟之前，需要調用tf.train.start_queue_runners函數，否則數據流圖將一直掛起。tf.train.start_queue_runners這個函數將會啟動輸入管道的線程，填充樣本到隊列中，以便出隊操作可以從隊列中拿到樣本。這種情況下最好配合使用一個tf.train.Coordinator,這樣可以再發生錯誤的情況下正確下正確地關閉這些線程。如果對訓練迭代數做了限制，那么需要使用一個訓練迭代數計數器，並且需要被初始化。

# Create the graph, etc.
init_op = tf.initialize_all_variables()  

# Create a session for running operations in the Graph.
sess = tf.Session()

# Initialize the variables (like the epoch counter).
sess.run(init_op)     #沒有初始化會報錯

# Start input enqueue threads.
coord = tf.train.Coordinator()
threads = tf.train.start_queue_runners(sess=sess, coord=coord)

try:
    while not coord.should_stop():
        # Run training steps or whatever
        sess.run(train_op)

except tf.errors.OutOfRangeError:
    print 'Done training -- epoch limit reached'
finally:
    # When done, ask the threads to stop.
    coord.request_stop()

# Wait for threads to finish.
coord.join(threads)
sess.close()

具體的數據輸入流程如下圖所示：

 先單個線程構建文件名隊列，然后用reader解析每個文件名中的數據，一個reader負責一個文件名中的數據，reader解析完數據樣本后，tf.train.shuffle_batch構建樣本隊列，對於單個reader就是按順序對每個文件中數據shuffle輸入隊列。多個reader就是同時對不同的文件中樣本數據shuffle輸入隊列。

因為開始就運行了這些入隊操作的線程，所以訓練循環會使得樣本隊列中中的樣本不斷地出隊。

在tf.train中要創建這些隊列和執行入隊操作，就要添加tf.train.QueueRunner到一個使用tf.train.add_queue_runner函數的數據流圖中。每個QueueRunner負責一個階段，處理那些需要在線程中運行的入隊操作的列表。一旦數據流圖構造成功，tf.train.start_queue_runners函數就會要求數據流圖中每個QueueRunner去開始它的線程運行入隊操作。

如果一切順利的話，你現在可以執行你的訓練步驟，同時隊列也會被后台線程來填充。如果您設置了最大訓練迭代數，在某些時候，樣本出隊的操作可能會得到一個tf.OutOfRangeError的錯誤。這其實是TensorFlow的“文件結束”（EOF） ———— 這就意味着已經達到了最大訓練迭代數，已經沒有更多可用的樣本了。

最后一個因素是Coordinator。這是負責在收到任何關閉信號的時候，讓所有的線程都知道。最常用的是在發生異常時這種情況就會呈現出來，比如說其中一個線程在運行某些操作時出現錯誤（或一個普通的Python異常）。

 

Queue

• Queue是TF隊列和緩存機制的實現
• QueueRunner是TF中對操作Queue的線程的封裝
• Coordinator是TF中用來協調線程運行的工具

Queue主要包含 入列（enqueue）和 出列（dequeue）兩個操作。enqueue操作返回計算圖中的一個Operation節點，dequeue操作返回一個Tensor值。Tensor在創建時同樣只是一個定義（或稱為“聲明”），需要放在Session中運行才能獲得真正的數值。

import tensorflow as tf
tf.InteractiveSession()

q = tf.FIFOQueue(2, "float")
init = q.enqueue_many(([0,0],))

x = q.dequeue()
y = x+1
q_inc = q.enqueue([y])

init.run()
q_inc.run()
q_inc.run()
q_inc.run()
x.eval()  # 返回1
x.eval()  # 返回2
x.eval()  # 卡住

 QueueRunner

Tensorflow的計算主要在使用CPU/GPU和內存，而數據讀取涉及磁盤操作，速度遠低於前者操作。因此通常會使用多個線程讀取數據，然后使用一個線程消費數據。QueueRunner就是來管理這些讀寫隊列的線程的。

QueueRunner需要與Queue一起使用（這名字已經注定了它和Queue脫不開干系），但並不一定必須使用Coordinator。

Coordinator

Coordinator是個用來保存線程組運行狀態的協調器對象，它和TensorFlow的Queue沒有必然關系，是可以單獨和Python線程使用的。

import tensorflow as tf
import threading, time

# 子線程函數
def loop(coord, id):
    t = 0
    while not coord.should_stop():
        print(id)
        time.sleep(1)
        t += 1
        # 只有1號線程調用request_stop方法
        if (t >= 2 and id == 1):
            coord.request_stop()

# 主線程
coord = tf.train.Coordinator()
# 使用Python API創建10個線程
threads = [threading.Thread(target=loop, args=(coord, i)) for i in range(10)]

# 啟動所有線程，並等待線程結束
for t in threads: t.start()
coord.join(threads)

 

將這個程序運行起來，會發現所有的子線程執行完兩個周期后都會停止，主線程會等待所有子線程都停止后結束，從而使整個程序結束。由此可見，只要有任何一個線程調用了Coordinator的request_stop方法，所有的線程都可以通過should_stop方法感知並停止當前線程。

將QueueRunner和Coordinator一起使用，實際上就是封裝了這個判斷操作，從而使任何一個現成出現異常時，能夠正常結束整個程序，同時主線程也可以直接調用request_stop方法來停止所有子線程的執行。

Queue的使用都是配合了QueueRunner和Coordinator一起使用的

• 第一種，顯式的創建QueueRunner，然后調用它的create_threads方法啟動線程。
• 第二種，使用全局的start_queue_runners方法啟動線程

在這個例子中，tf.train.string_input_produecer會將一個隱含的QueueRunner添加到全局圖中（類似的操作還有tf.train.shuffle_batch等）。

由於沒有顯式地返回QueueRunner來用create_threads啟動線程，這里使用了tf.train.start_queue_runners方法直接啟動tf.GraphKeys.QUEUE_RUNNERS集合中的所有隊列線程。這兩種方式在效果上是等效的。

 參考鏈接

http://wiki.jikexueyuan.com/project/tensorflow-zh/how_tos/reading_data.html#AUTOGENERATED-reading-from-files  官方文檔中文版

http://honggang.io/2016/08/19/tensorflow-data-reading/   解釋的相當清楚，還有具體例子說明 必看{❤❤}

https://saicoco.github.io/tf3/    解釋的相當棒

http://www.jianshu.com/p/d063804fb272    理解tensorflow中的Queue