讀取數(shù)據(jù)（Reading data）

TensorFlow輸入數(shù)據(jù)的方式有四種：

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

1. tf.data API

關(guān)于tf.data.Dataset的更詳盡解釋請看《》。tf.data API能夠從不同的輸入或文件格式中讀取、預(yù)處理數(shù)據(jù)，并且對數(shù)據(jù)應(yīng)用一些變換（例如，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 將讀取的內(nèi)容解析成一個Tensor列表。參數(shù) record_defaults 決定解析產(chǎn)生的Tensor的類型，另外，如果輸入中有缺失值，則用record_defaults 指定的默認(rèn)值來填充。

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

3.2.2 Fixed length records

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

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

3.2.3 Standard TensorFlow format

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

你寫一小段程序來獲取數(shù)據(jù)，然后將數(shù)據(jù)填入一個Example protocol buffer，并將這個 protocol buffer 序列化（serializes）為一個string，然后用 tf.python_io.TFRcordWriter 將這個string寫入到一個TFRecords文件中。例如，tensorflow/examples/how_tos/reading_data/convert_to_records.py 將MNIST數(shù)據(jù)集轉(zhuǎn)化為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)

為了完成相同的任務(wù)，基于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進(jìn)行你想要的預(yù)處理（preprocessing）。預(yù)處理是獨立的（不依賴于模型參數(shù)）。常見的預(yù)處理有：數(shù)據(jù)的標(biāo)準(zhǔn)化（normalization of your data）、挑選一個隨機的切片，添加噪聲（noise）或者畸變（distortions）等。具體的例子見：tensorflow_models/tutorials/image/cifar10/cifar10_input.py

3.4 Batching

在pipeline的末端，我們通過調(diào)用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讀取的內(nèi)容整合到一起。（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，讀取不同的文件，直到所有的文件的已經(jīng)讀取完，然后在下一個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.)）。

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

1. 如果你的num_threads參數(shù)值比文件的數(shù)量多，那么很有可能：有兩個threads會一前一后從同一個文件中讀取相同的example。這是不好的，應(yīng)該避免。
2. 或者，如果并行地讀取N個文件，可能或?qū)е麓罅康拇疟P搜索（意思是，多個文件存在于磁盤的不同位置，而磁頭只能有一個位置，所以會增加磁盤負(fù)擔(dān)）

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

3.5 Creating threads to prefetch using QueueRunner objects

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

說明：tf.train里的很多函數(shù)會添加tf.train.QueueRunner對象到你的graph。這些對象需要你在訓(xùn)練或者推理前，調(diào)用tf.train.start_queue_runners，否則數(shù)據(jù)無法讀取到圖中。調(diào)用tf.train.start_queue_runners會運行輸入pipeline需要的線程，這些線程將example enqueue到隊列中，然后dequeue操作才能成功。這最好和tf.train.Coordinator配合著用，當(dāng)有錯誤時，它會完全關(guān)閉掉開啟的threads。如果你在創(chuàng)建pipline時設(shè)置了迭代epoch數(shù)限制，將會創(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連在一起。第一個階段將會產(chǎn)生要讀取的文件的文件名，并將文件名enqueue到filename queue。第二個階段使用一個Reader來dequeue文件名并讀取，產(chǎn)生example，并將example enqueue到一個example queue。根據(jù)你的設(shè)置，你可能有很多第二階段（并行），所以你可以從并行地讀取多個文件。最后一個階段是一個enqueue操作，將example enqueue成一個queue，然后等待下一步操作。我們想要開啟多個線程運行著些enqueue操作，所以我們的訓(xùn)練loop能夠從example queue中dequeue examples。

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

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

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

關(guān)于threading、queues、QueueRunners、Coordinators的更多細(xì)節(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 設(shè)置為 0；如果你想讓每一個record產(chǎn)生多個example，你可以把batch size設(shè)置為大于1。然后，在調(diào)用調(diào)用batching函數(shù)（shuffle_batch或shuffle_batch_join）時，設(shè)置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

這僅僅適用于小數(shù)據(jù)集，小數(shù)據(jù)集可以被整體加載到內(nèi)存。預(yù)加載數(shù)據(jù)集主要有兩種方法：

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

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

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

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

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

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

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

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

4. Multiple input pipelines

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

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

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

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

你可以在同一個進(jìn)程中同一個圖中進(jìn)行訓(xùn)練和評估，并且訓(xùn)練和評估共享訓(xùn)練好的參數(shù)和層。關(guān)于共享變量，詳見這里。

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

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

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

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評論