聊一聊TensorFlow的數據導入機制

今天我們要講的是TensorFlow中的數據導入機制，傳統的做法是習慣于先構建好TF圖模型，然后開啟一個會話（Session），在運行圖模型之前將數據feed到圖中，這種做法的缺點是數據IO帶來的時間消耗很大，那么在訓練非常龐大的數據集的時候，不提倡采用這種做法，TensorFlow中取而代之的是tf.data.Dataset模塊，今天我們重點介紹這個。

tf.data是一個十分強大的可以用于構建復雜的數據導入機制的API，例如，如果你要處理的是圖像，那么tf.data可以幫助你把分布在不同位置的文件整合到一起，并且對每幅圖片添加微小的隨機噪聲，以及隨機選取一部分圖片作為一個batch進行訓練；又或者是你要處理文本，那么tf.data可以幫助從文本中解析符號并且轉換成embedding矩陣，然后將不同長度的序列變成一個個batch。

我們可以用tf.data.Dataset來構建一個數據集，數據集的來源可以有多種方式，例如如果你的數據集是預先以TFRecord格式寫在硬盤上的，那么你可以用tf.data.TFRecordDataset來構建；如果你的數據集是內存中的tensor變量，那么可以用tf.data.Dataset.from_tensors() 或 tf.data.Dataset.from_tensor_slices()來構建。下面我將通過代碼來演示它們。

首先，我們來看從內存中的tensor變量來構建數據集，如下代碼所示，首先構建了一個0～10的數據集，然后構建迭代器，迭代器可以每次從數據集中提取一個元素：

import tensorflow as tf dataset=tf.data.Dataset.range(10) iterator=dataset.make_one_shot_iterator() next_element = iterator.get_next()with tf.Session() as sess: for _ in range(10): print(sess.run(next_element))

如上代碼所示，range()是tf.data.Dataset類的一個靜態函數，用于產生一段序列。需要注意的是，構建的數據集需要是同一種數據類型以及內部結構。除此之外，由于range(10)代表0～9一共十個數，因此，這里的iterator只能運行10次，超過以后將會拋出tf.errors.OutOfRangeError異常。如果希望不拋出異常，則可以調用dataset.repeat(count)即可實現count次自動重復的迭代器。

range的范圍我們也可以在運行時才確定，即定義max_range為placeholder變量，這個時候需要調用Dataset的make_initializable_iterator方法來構建迭代器，并且這個迭代器的operation需要在迭代之前被運行，代碼如下所示：

max_range=tf.placeholder(tf.int64, shape=[]) dataset = tf.data.Dataset.range(max_range) iterator = dataset.make_initializable_iterator() next_element = iterator.get_next()with tf.Session() as sess: sess.run(iterator.initializer, feed_dict={max_range: 10}) for _ in range(10): print(sess.run(next_element))

也可以為不同的數據集創建同一個迭代器，為了使得這個迭代器可以被重復使用，需要保證不同數據集的類型和維度是一致的。例如，下面的代碼演示了如何使用同一個迭代器來構建訓練集和驗證集，可以看到，當我們開始訓練訓練集的時候，就需要先執行training_init_op，目的是使得迭代器開始加載訓練數據；而當進行驗證的時候，則需要先執行validation_init_op，道理一樣。

training_data = tf.data.Dataset.range(100).map(lambda x: x+tf.random_uniform([], -10, 10, tf.int64)) validation_data = tf.data.Dataset.range(50) iterator = tf.Iterator.from_structure(training_data.output_types, training_data.output_shapes) iterator = tf.data.Iterator.from_structure(training_data.output_types, training_data.output_shapes) next_element = iterator.get_next() training_init_op=iterator.make_initializer(training_data) validation_init_op=iterator.make_initializer(validation_data)with tf.Session() as sess: for epoch in range(10): sess.run(training_init_op) for _ in range(100): sess.run(next_element) sess.run(validation_init_op) for _ in range(50): sess.run(next_element)

也可以通過Tensor變量構建tf.data.Dataset，如下代碼所示，需要注意的是，這里的Tensor的維度是4×10，因此，傳入到迭代器中就是可以運行4次，每次運行生成一個長度為10的向量。

import tensorflow as tf dataset = tf.data.Dataset.from_tensor_slices(tf.random_uniform([4, 10])) iterator = dataset.make_initializable_iterator() next_element = iterator.get_next()with tf.Session() as sess: sess.run(iterator.initializer) for i in range(4): value = sess.run(next_element) print(value)

最后，還有一種比較常見的讀取數據的方式，就是從TFRecord文件中去讀取，這里再介紹一下之前在語音識別項目里采取的TFRecord的讀寫代碼。

首先是將音頻特征寫入到TFRecord文件之中，在語音識別中，我們最常用的兩個特征就是MFCC和LogFBank，要寫入文件中的不僅僅是這兩個變量，還要有文本標簽Label以及特征序列的長度sequence_legnth，這四個變量中，只有sequence_length是整數標量，其他三個都是列表格式，所以這里對于列表使用字節來保存，而對于標量，使用整型來保存。

def _bytes_feature(value): return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))def _int64_feature(value): return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))class RecordWriter(object): def __init__(self): pass def write(self, content, tfrecords_filename): writer = tf.python_io.TFRecordWriter(tfrecords_filename) if isinstance(content, list): feature_dict = {} for i in range(len(content)): feature = content[i] if i==0: feature_raw = np.array(feature).tostring() feature_dict['mfccFeat']=_bytes_feature(feature_raw) elif i==1: feature_raw = np.array(feature).tostring() feature_dict['logfbankFeat']=_bytes_feature(feature_raw) elif i==2: feature_raw = np.array(feature).tostring() feature_dict['label']=_bytes_feature(feature_raw) else: feature_dict['sequence_length']=_int64_feature(feature) features_to_write = tf.train.Example(features=tf.train.Features(feature=feature_dict)) writer.write(features_to_write.SerializeToString()) writer.close() print('Record has been writen:'+tfrecords_filename)

寫好TFRecord以后，在讀取的時候首先需要對TFRecord格式文件進行解析，解析函數如下：

def parse(self, serialized): feature_dict={} feature_dict['mfccFeat']=tf.FixedLenFeature([], tf.string) feature_dict['logfbankFeat']=tf.FixedLenFeature([], tf.string) feature_dict['label']=tf.FixedLenFeature([], tf.string) feature_dict['sequence_length']=tf.FixedLenFeature([1], tf.int64) features = tf.parse_single_example( serialized, features=feature_dict) mfcc = tf.reshape(tf.decode_raw(features['mfccFeat'], tf.float32), [-1, self.feature_num]) logfbank = tf.reshape(tf.decode_raw(features['logfbankFeat'], tf.float32), [-1, self.feature_num]) label = tf.decode_raw(features['label'], tf.int64) return mfcc, logfbank, label, features['sequence_length']

然后我們可以直接通過調用tf.data.TFRecordDataset來導入TFRecord文件列表，以及對每個文件調用parse函數進行解析，并且由于每個文件的特征矩陣長度不一，所以需要對齊進行padding操作，最終可以獲得迭代器，代碼如下：

self.fileNameList = tf.placeholder(tf.string, [None, ]) padded_shapes= ([-1,feature_num],[-1,feature_num],[-1],[1]) padded_values = (0.0,0.0,np.int64(-1),np.int64(0)) dataset = tf.data.TFRecordDataset(self.fileNameList, buffer_size=self.buffer_size).map(self.parse, num_parallel_call).padded_batch(batch_size, padded_shapes, padded_values) self.iterator = tf.data.Iterator.from_structure((tf.float32, tf.float32, tf.int64, tf.int64), (tf.TensorShape([None, None, 60]), tf.TensorShape([None, None, 60]), tf.TensorShape([None, None]), tf.TensorShape([None, None]))) self.initializer = self.iterator.make_initializer(dataset)

于是，關于TFRecord文件的讀寫就介紹完了，并且，基于TensorFlow的數據導入機制也介紹完了。