tensorflow讀取數據-tfrecord格式

01-29

歡迎關注我們的微信公眾號「人工智慧LeadAI」

概述

關於tensorflow讀取數據，官網給出了三種方法：

1、供給數據：在tensorflow程序運行的每一步，讓python代碼來供給數據

2、從文件讀取數據：建立輸入管線從文件中讀取數據

3、預載入數據：如果數據量不太大，可以在程序中定義常量或者變數來保存所有的數據。

這裡主要介紹一種比較通用、高效的數據讀取方法，就是tensorflow官方推薦的標準格式：tfrecord。

tfrecord數據文件

tfrecord數據文件是一種將圖像數據和標籤統一存儲的二進位文件，能更好的利用內存，在tensorflow中快速的複製，移動，讀取，存儲等。

tfrecord文件包含了tf.train.Example 協議緩衝區(protocol buffer，協議緩衝區包含了特徵 Features)。你可以寫一段代碼獲取你的數據，將數據填入到Example協議緩衝區(protocol buffer)，將協議緩衝區序列化為一個字元串，並且通過tf.python_io.TFRecordWriter class寫入到TFRecords文件。tensorflow/g3doc/how_tos/reading_data/convert_to_records.py就是這樣的一個例子。

tf.train.Example的定義如下：

message Example {n Features features = 1;n};nmessage Features{n map<string,Feature> featrue = 1;n};nmessage Feature{n oneof kind{n BytesList bytes_list = 1;n FloatList float_list = 2;n Int64List int64_list = 3;n }};n

從上述代碼可以看出，tf.train.Example中包含了屬性名稱到取值的字典，

其中屬性名稱為字元串，屬性的取值可以為字元串（BytesList）、實數列表（

FloatList）或者整數列表（Int64List）。

代碼實現

將數據保存為tfrecord格式

具體來說，首先需要給定tfrecord文件名稱，並創建一個文件：

tfrecords_filename = ./tfrecords/train.tfrecordswriter = tf.python_io.TFRecordWriter(tfrecords_filename) # 創建.tfrecord文件，準備寫入n

之後就可以創建一個循環來依次寫入數據：

for i in range(100):n img_raw = np.random.random_integers(0,255,size=(7,30)) # 創建7*30，取值在0-255之間隨機數組n img_raw = img_raw.tostring()n example = tf.train.Example(features=tf.train.Features(n feature={n label: tf.train.Feature(int64_list = tf.train.Int64List(value=[i])), n img_raw:tf.train.Feature(bytes_list = tf.train.BytesList(value=[img_raw]))n }))n writer.write(example.SerializeToString()) writer.close()n

example = tf.train.Example()這句將數據賦給了變數example（可以看到裡面是通過字典結構實現的賦值），然後用writer.write(example.SerializeToString()) 這句實現寫入。

值得注意的是賦值給example的數據格式。從前面tf.train.Example的定義可知，tfrecord支持整型、浮點數和二進位三種格式，分別是

tf.train.Feature(int64_list = tf.train.Int64List(value=[int_scalar]))ntf.train.Feature(bytes_list = tf.train.BytesList(value=[array_string_or_byte]))ntf.train.Feature(bytes_list = tf.train.FloatList(value=[float_scalar]))n

例如圖片等數組形式（array）的數據，可以保存為numpy array的格式，轉換為string，然後保存到二進位格式的feature中。對於單個的數值（scalar），可以直接賦值。這裡value=[×]的[]非常重要，也就是說輸入的必須是列表（list）。當然，對於輸入數據是向量形式的，可以根據數據類型（float還是int）分別保存。並且在保存的時候還可以指定數據的維數。

讀取tfrecord數據

從TFRecords文件中讀取數據，首先需要用tf.train.string_input_producer生成一個解析隊列。之後調用tf.TFRecordReader的tf.parse_single_example解析器。如下圖：

AnimatedFileQueues

解析器首先讀取解析隊列，返回serialized_example對象，之後調用tf.parse_single_example操作將Example協議緩衝區(protocol buffer)解析為張量。 MNIST的例子就使用了convert_to_records 所構建的數據。可以參考tensorflow/examples/how_tos/reading_data/fully_connected_reader.py。

下面代碼實現了tfrecord數據的讀取：

if __name__==』__main__』:ntfrecords_filename = "train.tfrecords"ntest_write_to_tfrecords(tfrecords_filename)nfilename_queue = tf.train.string_input_producer([tfrecords_filename],) #讀入流中nreader = tf.TFRecordReader()n_, serialized_example = reader.read(filename_queue) #返迴文件名和文件nfeatures = tf.parse_single_example(serialized_example,nfeatures={nlabel: tf.FixedLenFeature([], tf.int64),nimg_raw : tf.FixedLenFeature([], tf.string),n}) #取出包含image和label的feature對象nimage = tf.decode_raw(features[img_raw],tf.int64)nimage = tf.reshape(image, [7,30])nlabel = tf.cast(features[label], tf.int64)nwith tf.Session() as sess: #開始一個會話ninit_op = tf.initialize_all_variables()nsess.run(init_op)ncoord=tf.train.Coordinator()nthreads= tf.train.start_queue_runners(coord=coord)nfor i in range(20):nexample, l = sess.run([image,label])#在會話中取出image和labelnimg=Image.fromarray(example, RGB)#這裡Image是之前提到的nimg.save(./+str(i)+_Label_+str(l)+.jpg)#存下圖片nprint(example, l)nncoord.request_stop()ncoord.join(threads)n

程序中with tf.Session() as sess:創建了一個會話，其中特別要注意的就是以下兩句，非常重要

coord=tf.train.Coordinator() #創建一個協調器，管理線程nthreads= tf.train.start_queue_runners(coord=coord) #啟動QueueRunner, 此時文件名隊列已經進隊。n

這兩句實現的功能就是創建線程並使用QueueRunner對象來提取數據。簡單來說：使用tf.train函數添加QueueRunner到tensorflow中。在運行任何訓練步驟之前，需要調用tf.train.start_queue_runners函數，否則tensorflow將一直掛起。

tf.train.start_queue_runners 這個函數將會啟動輸入管道的線程，填充樣本到隊列中，以便出隊操作可以從隊列中拿到樣本。這種情況下最好配合使用一個tf.train.Coordinator，這樣可以在發生錯誤的情況下正確地關閉這些線程。如果你對訓練迭代數做了限制，那麼需要使用一個訓練迭代數計數器，並且需要被初始化。詳細解釋可以參考鏈接[4]。

推薦的代碼模板如下：

# Create the graph, etc.ninit_op = tf.initialize_all_variables()nn# Create a session for running operations in the Graph.nsess = tf.Session()nn# Initialize the variables (like the epoch counter).nsess.run(init_op)nn# Start input enqueue threads.ncoord = tf.train.Coordinator()nthreads = tf.train.start_queue_runners(sess=sess, coord=coord)nntry:n while not coord.should_stop():n # Run training steps or whatevern sess.run(train_op)nnexcept tf.errors.OutOfRangeError:n print Done training -- epoch limit reachednfinally:n # When done, ask the threads to stop.n coord.request_stop()nn# Wait for threads to finish.ncoord.join(threads)nsess.close()n

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

在執行訓練的時候，隊列會被後台的線程填充好。如果設置了最大訓練迭代數（epoch），在某些時候，樣本出隊的操作可能會拋出一個tf.OutOfRangeError的錯誤。這是因為tensorflow的隊列已經到達了最大實際的最大迭代數，沒有更多可用的樣本了。這也是為何推薦代碼模板需要用try..except ..finally結構來處理這種錯誤。

上面代碼讀取的是單個的image和label，而在tensorflow訓練時，一般是採取batch的方式去讀入數據。tensorflow提供了兩種方式，一種是shuffle_batch（tf.train.shuffle_batch），這種主要是用在訓練中，隨機選取樣本組成batch。另外一種就是按照數據在tfrecord中的先後順序生成batch（tf.train.batch）。

這裡採用tf.train.shuffle_batch方式：

# Creates batches of 32 images and 32 labels.nimage_batch, label_batch = tf.train.shuffle_batch(n[image, label],nbatch_size=3,nnum_threads=4,ncapacity=10,nmin_after_dequeue=10)n

tf.train.shuffle_batch函數輸入參數為：

tensor_list: 進入隊列的張量列表The list of tensors to enqueue.
batch_size: 從數據隊列中抽取一個批次所包含的數據條數The new batch size pulled from the queue.
capacity: 隊列中最大的數據條數An integer. The maximum number of elements in the queue.
min_after_dequeue: 提出隊列後，隊列中剩餘的最小數據條數Minimum number elements in the queue after a dequeue, used to ensure a level of mixing of elements.
num_threads: 進行隊列操作的線程數目The number of threads enqueuing tensor_list.
seed: 隊列中進行隨機排列的隨機數發生器，似乎不常用到Seed for the random shuffling within the queue.
enqueue_many: 張量列表中的每個張量是否是一個單獨的例子，似乎不常用到Whether each tensor in tensor_list is a single example.
shapes: (Optional) The shapes for each example. Defaults to the inferred shapes for tensor_list.
name: (Optional) A name for the operations.

值得注意的是，capacity>=min_after_dequeue+num_threads*batch_size。

最後附上完整代碼：

import tensorflow as tfnimport numpy as npnimport osn#=============================================================================#n# write images and label in tfrecord file and read them outndef encode_to_tfrecords(tfrecords_filename, data_num): n write into tfrecord file nif os.path.exists(tfrecords_filename):n os.remove(tfrecords_filename)nnwriter = tf.python_io.TFRecordWriter(./+tfrecords_filename) # 創建.tfrecord文件，準備寫入nnfor i in range(data_num):nimg_raw = np.random.randint(0,255,size=(56,56))nimg_raw = img_raw.tostring()nexample = tf.train.Example(features=tf.train.Features(nfeature={nlabel: tf.train.Feature(int64_list = tf.train.Int64List(value=[i])), n img_raw:tf.train.Feature(bytes_list = tf.train.BytesList(value=[img_raw]))n}))nwriter.write(example.SerializeToString()) nnwriter.close()n return 0nndef decode_from_tfrecords(filename_queue, is_batch):nnreader = tf.TFRecordReader()n _, serialized_example = reader.read(filename_queue) #返迴文件名和文件nfeatures = tf.parse_single_example(serialized_example,n features={n label: tf.FixedLenFeature([], tf.int64),n img_raw : tf.FixedLenFeature([], tf.string),n }) #取出包含image和label的feature對象nimage = tf.decode_raw(features[img_raw],tf.int64)nimage = tf.reshape(image, [56,56])nlabel = tf.cast(features[label], tf.int64)nnif is_batch:n batch_size = 3nmin_after_dequeue = 10ncapacity = min_after_dequeue+3*batch_sizenimage, label = tf.train.shuffle_batch([image, label],nbatch_size=batch_size, nnum_threads=3, n capacity=capacity,nmin_after_dequeue=min_after_dequeue)nreturn image, labelnn#=============================================================================#nnif __name__==__main__:n # make train.tfrecordntrain_filename = "train.tfrecords"nencode_to_tfrecords(train_filename,100)n## # make test.tfrecordntest_filename = test.tfrecordsnencode_to_tfrecords(test_filename,10)nn# run_test = Truenfilename_queue = tf.train.string_input_producer([train_filename],num_epochs=None) #讀入流中ntrain_image, train_label = decode_from_tfrecords(filename_queue, is_batch=True)nnfilename_queue = tf.train.string_input_producer([test_filename],num_epochs=None) #讀入流中ntest_image, test_label = decode_from_tfrecords(filename_queue, is_batch=True)n with tf.Session() as sess: #開始一個會話ninit_op = tf.global_variables_initializer()nsess.run(init_op)ncoord=tf.train.Coordinator()nthreads= tf.train.start_queue_runners(coord=coord)nntry:n# while not coord.should_stop():nfor i in range(2):nexample, l = sess.run([train_image,train_label])#在會話中取出image和labelnprint(train:)nprint(example, l) ntexample, tl = sess.run([test_image, test_label])n print(test:)nprint(texample,tl)nexcept tf.errors.OutOfRangeError:nprint(Done reading)nfinally:ncoord.request_stop()nncoord.request_stop()ncoord.join(threads)n

另一段tfrecord代碼，這段代碼實現了float，int和string三種類型數據tfrecord

格式的編碼和解碼。特別注意的是，這裡編碼和解碼時，指定了數據的維度。

#!/usr/bin/env python3n# -*- coding: utf-8 -*-n"""nCreated on Mon Jul 24 15:47:28 2017nn@author: dln"""nimport tensorflow as tfnimport numpy as npnnwriter = tf.python_io.TFRecordWriter(test.tfrecord)nnfor i in range(0, 2):n a = np.random.random(size=(180)).astype(np.float32)n a = a.data.tolist()n b = [2016 + i, 2017+i]n c = np.array([[0, 1, 2],[3, 4, 5]]) + in c = c.astype(np.uint8)n c_raw = c.tostring()#這裡是把ｃ換了一種格式存儲n print( i:, i)n print( a:, a)n print( b:, b)n print( c:, c)n example = tf.train.Example(features=tf.train.Features(n feature = {a:tf.train.Feature(float_list = tf.train.FloatList(value=a)),n b:tf.train.Feature(int64_list = tf.train.Int64List(value = b)),n c:tf.train.Feature(bytes_list = tf.train.BytesList(value = [c_raw]))}))nserialized = example.SerializeToString()n writer.write(serialized)n print( writer,i,DOWN!)nwriter.close()nnnfilename_queue = tf.train.string_input_producer([test.tfrecord], num_epochs=None)n# create a reader from file queuenreader = tf.TFRecordReader()n_, serialized_example = reader.read(filename_queue)n# get feature from serialized examplennfeatures = tf.parse_single_example(serialized_example,n features={n a: tf.FixedLenFeature([180], tf.float32),nb: tf.FixedLenFeature([2], tf.int64),n c: tf.FixedLenFeature([],tf.string)n }n )na_out = features[a]nb_out = features[b]nc_out = features[c]n#c_raw_out = features[c]n#c_raw_out = tf.sparse_to_dense(features[c])n#c_out = tf.decode_raw(c_raw_out, tf.uint8)nprint( a_out)nprint( b_out)nprint( c_out)nna_batch, b_batch, c_batch = tf.train.shuffle_batch([a_out, b_out, c_out], batch_size=3, n capacity=200, min_after_dequeue=100, num_threads=2)nnnprint( a_batch)nprint( b_batch)nprint( c_batch)n

附：tensorflowd協調器和隊列運行器(Coordinator and QueueRunner)

參考

[1]http://www.2cto.com/kf/201702/604326.html

[2]這篇特別詳細

[3]http://blog.csdn.net/lujiandong1/article/details/53376802

[4]http://wiki.jikexueyuan.com/project/tensorflow-zh/how_tos/reading_data.html

[5]http://www.cnblogs.com/upright/p/6136265.html