讀取數據（Reading data）

TensorFlow輸入數據的方式有四種：

1. tf.data API：可以很容易的構建一個復雜的輸入通道(pipeline)（首選數據輸入方式）（Eager模式必須使用該API來構建輸入通道）
2. Feeding：使用Python代碼提供數據，然后將數據feeding到計算圖中。
3. QueueRunner：基于隊列的輸入通道（在計算圖計算前從隊列中讀取數據）
4. Preloaded data：用一個constant常量將數據集加載到計算圖中（主要用于小數據集）

1. tf.data API

關于tf.data.Dataset的更詳盡解釋請看《》。tf.data API能夠從不同的輸入或文件格式中讀取、預處理數據，并且對數據應用一些變換（例如，batching、shuffling、mapping function over the dataset），tf.data API 是舊的 feeding、QueueRunner的升級。

with tf.Session():
 input = tf.placeholder(tf.float32)
 classifier = ...
 print(classifier.eval(feed_dict={input: my_python_preprocessing_fn()}))

為了讀取逗號分隔符分割的text文件（csv），要使用一個 tf.TextLineReader 和一個 tf.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.stack([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 方法每執(zhí)行一次，會從文件中讀取一行。然后 decode_csv 將讀取的內容解析成一個Tensor列表。參數 record_defaults 決定解析產生的Tensor的類型，另外，如果輸入中有缺失值，則用record_defaults 指定的默認值來填充。

在使用run或者eval 執(zhí)行 read 方法前，你必須調用 tf.train.start_queue_runners 去填充 queue。否則，read 方法將會堵塞（等待 filenames queue 中 enqueue 文件名）。

3.2.2 Fixed length records

為了讀取二進制文件（二進制文件中，每一個record都占固定bytes），需要使用一個 tf.FixedLengthRecordReader 和 tf.decode_raw。decode_raw 將 reader 讀取的 string 解析成一個uint8 tensor。

例如，二進制格式的CIFAR-10數據集中的每一個record都占固定bytes：label占1 bytes，然后后面的image數據占3072 bytes。當你有一個unit8 tensor時，通過切片便可以得到各部分并reformat成需要的格式。對于CIFAR-10數據集的reading和decoding，可以參照：tensorflow_models/tutorials/image/cifar10/cifar10_input.py或這個教程。

3.2.3 Standard TensorFlow format

另一個方法是將數據集轉換為一個支持的格式。這個方法使得數據集和網絡的混合和匹配變得簡單（make it easier to mix and match data sets and network architectures）。TensorFlow中推薦的格式是 TFRecords文件，TFRecords中包含 tf.train.Example protocol buffers (在這個協(xié)議下，特征是一個字段).

你寫一小段程序來獲取數據，然后將數據填入一個Example protocol buffer，并將這個 protocol buffer 序列化（serializes）為一個string，然后用 tf.python_io.TFRcordWriter 將這個string寫入到一個TFRecords文件中。例如，tensorflow/examples/how_tos/reading_data/convert_to_records.py 將MNIST數據集轉化為TFRecord格式。

讀取TFRecord文件的推薦方式是使用 tf.data.TFRecordDataset，像這個例子一樣:

dataset = tf.data.TFRecordDataset(filename)
dataset = dataset.repeat(num_epochs)

# map takes a python function and applies it to every sample
dataset = dataset.map(decode)

為了完成相同的任務，基于queue的輸入通道需要下面的代碼（使用的decode和上一段代碼一樣）：

filename_queue = tf.train.string_input_producer([filename], num_epochs=num_epochs)
reader = tf.TFRecordReader()
_, serialized_example = reader.read(filename_queue)
image,label = decode(serialized_example)

3.3 Preprocessing

然后你可以對examples進行你想要的預處理（preprocessing）。預處理是獨立的（不依賴于模型參數）。常見的預處理有：數據的標準化（normalization of your data）、挑選一個隨機的切片，添加噪聲（noise）或者畸變（distortions）等。具體的例子見：tensorflow_models/tutorials/image/cifar10/cifar10_input.py

3.4 Batching

在pipeline的末端，我們通過調用tf.train.shuffle_batch 來創(chuàng)建兩個queue，一個將example batch起來 for training、evaluation、inference；另一個來shuffle examples的順序。

例子：

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)
 # 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

如果你需要更多的并行或者打亂不同文件中example，使用多個reader，然后使用 tf.train.shuffle_batch_join將多個reader讀取的內容整合到一起。（If you need more parallelism or shuffling of examples between files, use multiple reader instances using the tf.train.shuffle_batch_join）

例子：

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, read_threads, num_epochs=None):
 filename_queue = tf.train.string_input_producer(
   filenames, num_epochs=num_epochs, shuffle=True)
 example_list = [read_my_file_format(filename_queue)
         for _ in range(read_threads)]
 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

所有的reader共享一個filename queue。這種方式保證了不同的reader在同一個epoch，讀取不同的文件，直到所有的文件的已經讀取完，然后在下一個epoch，重新從所有的文件讀?。╕ou still only use a single filename queue that is shared by all the readers. That way we ensure that the different readers use different files from the same epoch until all the files from the epoch have been started. (It is also usually sufficient to have a single thread filling the filename queue.)）。

另一個可選的方法是去通過調用 tf.train.shuffle_batch 使用單個的reader，但是將參數 num_threads 參數設置為大于1的值。這將使得在同一時間只能從一個文件讀取內容（但是比 1 線程快），而不是同時從N個文件中讀取。這可能很重要：

1. 如果你的num_threads參數值比文件的數量多，那么很有可能：有兩個threads會一前一后從同一個文件中讀取相同的example。這是不好的，應該避免。
2. 或者，如果并行地讀取N個文件，可能或導致大量的磁盤搜索（意思是，多個文件存在于磁盤的不同位置，而磁頭只能有一個位置，所以會增加磁盤負擔）

那么需要多少個線程呢？tf.train.shuffle_batch*函數會給計算圖添加一個summary來記錄 example queue 的使用情況。如果你有足夠的reading threads，這個summary將會總大于0。你可以用TensorBoard來查看訓練過程中的summaries

3.5 Creating threads to prefetch using QueueRunner objects

使用QueueRunner對象來創(chuàng)建threads來prefetch數據

說明：tf.train里的很多函數會添加tf.train.QueueRunner對象到你的graph。這些對象需要你在訓練或者推理前，調用tf.train.start_queue_runners，否則數據無法讀取到圖中。調用tf.train.start_queue_runners會運行輸入pipeline需要的線程，這些線程將example enqueue到隊列中，然后dequeue操作才能成功。這最好和tf.train.Coordinator配合著用，當有錯誤時，它會完全關閉掉開啟的threads。如果你在創(chuàng)建pipline時設置了迭代epoch數限制，將會創(chuàng)建一個epoch counter的局部變量（需要初始化）。下面是推薦的代碼使用模板：

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

# 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()

這里的代碼是怎么工作的？

首先，我們創(chuàng)建整個圖。它的input pipeline將有幾個階段，這些階段通過Queue連在一起。第一個階段將會產生要讀取的文件的文件名，并將文件名enqueue到filename queue。第二個階段使用一個Reader來dequeue文件名并讀取，產生example，并將example enqueue到一個example queue。根據你的設置，你可能有很多第二階段（并行），所以你可以從并行地讀取多個文件。最后一個階段是一個enqueue操作，將example enqueue成一個queue，然后等待下一步操作。我們想要開啟多個線程運行著些enqueue操作，所以我們的訓練loop能夠從example queue中dequeue examples。

tf.train里的輔助函數（創(chuàng)建了這些queue、enqueuing操作）會調用tf.train.add_queue——runner添加一個tf.train.QueueRunner到圖中。每一個QueueRunner負責一個階段。一旦圖構建好，tf.train.start_queue_runners函數會開始圖中每一個QueueRunner的入隊操作。

如果一切進行順利，你現在可以運行訓練step（后臺線程會填滿queue）。如果你設置了epoch限制，在達到固定的epoch時，在進行dequeuing會得到tf.errors.OutOfRangeError。這個錯誤等價于EOF（end of file），意味著已經達到了固定的epochs。

最后一部分是tf.train.Coordinator。它主要負責通知所有的線程是否應該停止。在大多數情況下，這通常是因為遇到了一個異常（exception）。例如，某一個線程在運行某些操作時出錯了（或者python的異常）。

關于threading、queues、QueueRunners、Coordinators的更多細節(jié)見這里

3.6 Filtering records or producing multiple examples per record

一個example的shape是 [x,y,z]，一個batch的example的shape為 [batch, x, y, z]。如果你想去過濾掉這個record，你可以把 batch size 設置為 0；如果你想讓每一個record產生多個example，你可以把batch size設置為大于1。然后，在調用調用batching函數（shuffle_batch或shuffle_batch_join）時，設置enqueue_many=True。

3.7 Sparse input data

queues在SparseTensors的情況下不能很好的工作。如果你使用SparseTensors，你必須在batching后用tf.sparse_example來decode string records（而不是在batching前使用tf.parse_single_example來decode）

4. Preloaded data

這僅僅適用于小數據集，小數據集可以被整體加載到內存。預加載數據集主要有兩種方法：

1. 將數據集存儲成一個constant
2. 將數據集存儲在一個variable中，一旦初始化或者assign to后，便不再改變。

使用一個constant更簡單，但是需要更多的內存（因為所有的常量都儲存在計算圖中，而計算圖可能需要進行多次復制）。

training_data = ...
training_labels = ...
with tf.Session():
 input_data = tf.constant(training_data)
 input_labels = tf.constant(training_labels)
 ...

為了使用一個varibale，在圖構建好后，你需要去初始化它。

training_data = ...
training_labels = ...
with tf.Session() as sess:
 data_initializer = tf.placeholder(dtype=training_data.dtype,
                  shape=training_data.shape)
 label_initializer = tf.placeholder(dtype=training_labels.dtype,
                   shape=training_labels.shape)
 input_data = tf.Variable(data_initializer, trainable=False, collections=[])
 input_labels = tf.Variable(label_initializer, trainable=False, collections=[])
 ...
 sess.run(input_data.initializer,
      feed_dict={data_initializer: training_data})
 sess.run(input_labels.initializer,
      feed_dict={label_initializer: training_labels})

設置trainable=False將使variable不加入GraphKeys.TRAINABLE_VARIABLES容器，所以我們不用在訓練過程中更新它。設置collections=[]將會使variable不加入GraphKeys.GLOBAL_VARIABLES容器（這個容器主要用于保存和恢復checkpoints）。

無論哪種方式，tf.train.slice_input_producer都能夠用來產生一個slice。這在整個epoch上shuffle了example，所以batching時，進一步的shuffling不再需要。所以不再使用shuffle_batch函數，而使用tf.train.batch函數。為了使用多個預處理線程，設置num_threads參數大于1。

MNIST數據集上使用constant來preload數據的實例見tensorflow/examples/how_tos/reading_data/fully_connected_preloaded.py；使用variable來preload數據的例子見tensorflow/examples/how_tos/reading_data/fully_connected_preloaded_var.py，你可以通過 fully_connected_feed和 fully_connected_feed版本來對比兩種方式。

4. Multiple input pipelines

一般，你想要去在一個數據集上訓練，而在另一個數據集上評估模型。實現這個想法的一種方式是：以兩個進程，建兩個獨立的圖和session：

1. 訓練進程讀取訓練數據，并且周期性地將模型的所有訓練好的變量保存到checkpoint文件中。
2. 評估進程從checkpoint文件中恢復得到一個inference模型，這個模型讀取評估數據。

在estimators里和CIFAR-10模型示例里，采用就是上面的方法。該方法主要有兩個好處：

1. 你的評估是在一個訓練好的模型的快照上進行的。
2. 在訓練完成或中斷后，你也可以進行評估。

你可以在同一個進程中同一個圖中進行訓練和評估，并且訓練和評估共享訓練好的參數和層。關于共享變量，詳見這里。

為了支持單個圖方法（single-graph approach），tf.data也提供了高級的iterator類型，它將允許用戶去在不重新構建graph和session的情況下，改變輸入pipeline。

注意：盡管上面的實現很好，但很多op（比如tf.layers.batch_normalization和tf.layers.dropout）與模型模式有關（訓練和評估時，計算不一致），你必須很小心地去設置這些，如果你更改數據源。

英文版：https://tensorflow.google.cn/api_guides/python/reading_data#_tf_data_API

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

最新評論