一、隊列和線程
1、隊列:
1)、tf.FIFOQueue(capacity, dtypes, name='fifo_queue') 創建一個以先進先出的順序對元素進行排隊的隊列
參數:
capacity：整數。可能存儲在此隊列中的元素數量的上限
dtypes：DType對象列表。長度dtypes必須等於每個隊列元 素中的張量數,dtype的類型形狀，決定了后面進隊列元素形狀
方法:
q.dequeue()獲取隊列的數據
q.enqueue(值)將一個數據添加進隊列
q.enqueue_many(列表或者元組)將多個數據添加進隊列
q.size() 返回隊列的大小
2)、tf.RandomShuffleQueue() 隨機出的隊列
2、隊列管理器
tf.train.QueueRunner(queue, enqueue_ops=None)
參數:
queue：A Queue
enqueue_ops：添加線程的隊列操作列表，[]*2,指定兩個線程
create_threads(sess, coord=None,start=False) 創建線程來運行給定會話的入隊操作
start：布爾值，如果True啟動線程；如果為False調用者 必須調用start()啟動線程
coord:線程協調器 用於線程的管理
3、線程協調器
tf.train.Coordinator() 線程協調員,實現一個簡單的機制來協調一 組線程的終止
方法: 返回的是線程協調實例
request_stop() 請求停止
join(threads=None, stop_grace_period_secs=120) 等待線程終止
結合隊列、隊列管理器和線程協調器實現異步的小例:
import tensorflow as tf # 1.創建隊列 Q = tf.FIFOQueue(2000, tf.float32) # 2.添加數據進隊列 # 2.1創建一個數據(變量) var = tf.Variable(0.0, tf.float32) # 2.2數據自增 plus = tf.assign_add(var, 1) # 2.3將數據添加進隊列 en_q = Q.enqueue(plus) # 3.創建隊列管理器 qr = tf.train.QueueRunner(Q, enqueue_ops=[en_q] * 2) # 4.變量初始化 init = tf.global_variables_initializer() # 5.創建會話 with tf.Session() as sess: # 6.運行初始化 sess.run(init) # 7.創建線程協調器 coord = tf.train.Coordinator() # 8.開啟子線程 threads = qr.create_threads(sess, coord=coord, start=True) # 9.主線程 從隊列中取數據 for i in range(200): print(sess.run(Q.dequeue())) # 10.線程回收 coord.request_stop() coord.join(threads)
二、文件讀取
1、文件讀取流程
2、文件讀取API
1)文件隊列
tf.train.string_input_producer(string_tensor, ,shuffle=True) 將輸出字符串（例如文件名）輸入到管道隊列
參數:
string_tensor 含有文件名的1階張量
num_epochs:過幾遍數據，默認無限過數據
返回:具有輸出字符串的隊列
2)文件閱讀器(根據文件格式，選擇對應的文件閱讀器)
csv文件: class tf.TextLineReader 默認按行讀取 返回：讀取器實例
二進制文件: tf.FixedLengthRecordReader(record_bytes) record_bytes:整型，指定每次讀取的字節數 返回：讀取器實例
TfRecords文件: tf.TFRecordReader 返回：讀取器實例
以上3個閱讀器有一個相同的方法:
read(file_queue)：從隊列中指定數量內容 返回一個Tensors元組（key, value） 其中key是文件名字，value是默認的內容(行，字節)
3)文件內容解碼器(由於從文件中讀取的是字符串，需要函數去解析這些字符串到張量)
①tf.decode_csv(records,record_defaults=None,field_delim = None，name = None) 將CSV轉換為張量，與tf.TextLineReader搭配使用
參數:
records:tensor型字符串，每個字符串是csv中的記錄行
field_delim:默認分割符”,”
record_defaults:參數決定了所得張量的類型，並設置一個值在輸入字符串中缺少使用默認值
②tf.decode_raw(bytes,out_type,little_endian = None，name = None) 將字節轉換為一個數字向量表示，字節為一字符串類型的張量,與函數tf.FixedLengthRecordReader搭配使用，二進制讀取為uint8格式
4)開啟線程操作
tf.train.start_queue_runners(sess=None,coord=None) 收集所有圖中的隊列線程，並啟動線程 sess:所在的會話中 coord：線程協調器 return：返回所有線程隊列
5)管道讀端批處理
①tf.train.batch(tensors,batch_size,num_threads = 1,capacity = 32,name=None) 讀取指定大小（個數）的張量
參數:
tensors：可以是包含張量的列表
batch_size:從隊列中讀取的批處理大小
num_threads：進入隊列的線程數
capacity：整數，隊列中元素的最大數量
返回:tensors
②tf.train.shuffle_batch(tensors,batch_size,capacity,min_after_dequeue, num_threads=1,) 亂序讀取指定大小（個數）的張量
參數:
min_after_dequeue:留下隊列里的張量個數，能夠保持隨機打亂
3、文件讀取案例
import tensorflow as tf import os def csv_read(filelist): # 構建文件隊列 Q = tf.train.string_input_producer(filelist) # 構建讀取器 reader = tf.TextLineReader() # 讀取隊列 key, value = reader.read(Q) # 構建解碼器 x1, y = tf.decode_csv(value, record_defaults=[["None"], ["None"]]) # 進行管道批處理 x1_batch, y_batch = tf.train.batch([x1, y], batch_size=12, num_threads=1, capacity=12) # 開啟會話 with tf.Session() as sess: # 創建線程協調器 coord = tf.train.Coordinator() # 開啟線程 threads = tf.train.start_queue_runners(sess, coord=coord) # 執行任務 print(sess.run([x1_batch, y_batch])) # 線程回收 coord.request_stop() coord.join(threads) if __name__ == "__main__": filename = os.listdir("./data/") # 文件目錄自己指定 filelist = [os.path.join("./data/", file) for file in filename] csv_read(filelist)
三、圖片讀取與存儲
1 圖像數字化三要素:長度,寬度,通道數(一通道 : 灰度值 三通道 : RGB)
2 縮小圖片大小:
tf.image.resize_images(images, size) 縮小圖片
目的：
1、增加圖片數據的統一性
2、所有圖片轉換成指定大小
3、縮小圖片數據量，防止增加開銷
3 圖像讀取API
1)圖像讀取器
tf.WholeFileReader 將文件的全部內容作為值輸出的讀取器
return：讀取器實例 read(file_queue):輸出將是一個文件名（key）和該文件的內容 （值）
2)圖像解碼器
tf.image.decode_jpeg(contents) 將JPEG編碼的圖像解碼為uint8張量
return:uint8張量，3-D形狀[height, width, channels]
tf.image.decode_png(contents) 將PNG編碼的圖像解碼為uint8或uint16張量
return:張量類型，3-D形狀[height, width, channels]
圖片讀取案的簡單demo:
import tensorflow as tf import os flags = tf.app.flags.FLAGS tf.app.flags.DEFINE_string("data_home", "./data/dog/", "狗的圖片目錄") # 文件路徑自己指定 def picread(filelist): # 構建文件名隊列 file_q = tf.train.string_input_producer(filelist) # 構建讀取器 reader = tf.WholeFileReader() # 讀取內容 key, value = reader.read(file_q) print(value) # 構建解碼器 image = tf.image.decode_jpeg(value) print(image) # 統一圖片大小 設置長寬 resize_image = tf.image.resize_images(image, [256,256]) print(resize_image) # 指定通道大小 resize_image.set_shape([256,256,3]) # 構建批量處理管道 image_batch = tf.train.batch([resize_image], batch_size=100,num_threads=1, capacity=100) return image_batch if __name__ == "__main__": filename = os.listdir(flags.data_home) filelist = [os.path.join(flags.data_home, file) for file in filename] image_batch = picread(filelist) with tf.Session() as sess: # 構建線程協調器 coord = tf.train.Coordinator() # 開啟線程 threads = tf.train.start_queue_runners(sess,coord=coord) # 訓練數據 print(sess.run(image_batch)) # 回收線程 coord.request_stop() coord.join(threads)
四、TFRecords分析、存取
1 概念
TFRecords是Tensorflow設計的一種內置文件格式，是一種二進制文件， 它能更好的利用內存，更方便復制和移動 (將二進制數據和標簽(訓練的類別標簽)數據存儲在同一個文件中)
2 TFRecords文件分析
1)文件格式：*.tfrecords
2)寫入文件內容：Example協議塊
3 TFRecords存儲
1)建立TFRecord存儲器
tf.python_io.TFRecordWriter(path) 寫入tfrecords文件
參數:
path: TFRecords文件的路徑
return：無, 執行寫文件操作
方法:
write(record):向文件中寫入一個字符串記錄 # 一個序列化的Example,Example.SerializeToString()
close():關閉文件寫入器
2)構造每個樣本的Example協議塊
tf.train.Example(features=None) 寫入tfrecords文件
參數:
features:tf.train.Features類型的特征實例
return：example格式協議塊
tf.train.Features(feature=None) 構建每個樣本的信息鍵值對
參數:
feature:字典數據,key為要保存的名字
value為tf.train.Feature實例
return:Features類型
tf.train.Feature(**options)
參數:
**options：例如 bytes_list=tf.train. BytesList(value=[Bytes])
int64_list=tf.train. Int64List(value=[Value])
float_list = tf.train. FloatList(value=[value])
4 TFRecords讀取方法
1)構建文件隊列
tf.train.string_input_producer(string_tensor, ,shuffle=True)
2)構建文件讀取器,讀取隊列的數據
tf.TFRecordReader 返回：讀取器實例
read(file_queue)
3)解析TFRecords的example協議內存塊
①tf.parse_single_example(serialized,features=None,name=None) 解析一個單一的Example原型
參數:
serialized：標量字符串Tensor，一個序列化的Example
features：dict字典數據，鍵為讀取的名字，值為FixedLenFeature
return:一個鍵值對組成的字典，鍵為讀取的名字
②tf.FixedLenFeature(shape,dtype)
參數:
shape：輸入數據的形狀，一般不指定,為空列表
dtype：輸入數據類型，與存儲進文件的類型要一致 類型只能是float32,int64,string
4)解碼
tf.decode_raw(bytes,out_type,little_endian = None，name = None) 將字節轉換為一個數字向量表示，字節為一字符串類型的張量,與函數tf.FixedLengthRecordReader搭配使用，二進制讀取為uint8格式
以下是從二進制文件中讀取數據,寫入tfrecords文件,再從tfrecords文件讀取的小案例:
import tensorflow as tf import os flags = tf.app.flags.FLAGS
tf.app.flags.DEFINE_string("data_home", "./data/cifar10/cifar-10-batches-bin/", "二進制文件目錄") tf.app.flags.DEFINE_string("data_tfrecords", "./data/temp/tfrecords", "tfrecords文件路徑") class cifarread(object): def __init__(self, filelist): self.filelist = filelist # 構建圖的一些數據 self.height = 32 self.width = 32 self.channel = 3 self.label_bytes = 1 self.image_bytes = self.height * self.width*self.channel self.bytes = self.label_bytes + self.image_bytes def read_decode(self): """ 讀取二進制文件 :return: image_batch, label_batch """ # 構建文件名隊列 file_q = tf.train.string_input_producer(self.filelist) # 構建閱讀器 reader = tf.FixedLengthRecordReader(record_bytes=self.bytes) # 讀取數據 key, value = reader.read(file_q) # 解碼 label_image = tf.decode_raw(value, tf.uint8) # 分割數據集 label = tf.cast(tf.slice(label_image, [0], [self.label_bytes]), tf.int32) image = tf.slice(label_image, [self.label_bytes], [self.image_bytes]) # 改變形狀 image_tensor = tf.reshape(image, [self.height, self.width, self.channel]) # 批量處理 image_batch, label_batch = tf.train.batch([image_tensor, label], batch_size=10, num_threads=1, capacity=10) return image_batch, label_batch def write2tfrecords(self, image_batch, label_batch): """ 將從二進制文件中讀取的內容寫入tfrecords文件 :param image_batch: :param label_batch: :return: """ # 構建一個tfrecords文件存儲器 writer = tf.python_io.TFRecordWriter(flags.data_tfrecords) # 對於每一個樣本，都要構造example寫入 for i in range(10): # 取出特征值，轉換成字符串 image_string = image_batch[i].eval().tostring() # 取出目標值 label_int = int(label_batch[i].eval()[0]) example = tf.train.Example(features=tf.train.Features(feature={ "image":tf.train.Feature(bytes_list = tf.train.BytesList(value=[image_string])), "label":tf.train.Feature(int64_list = tf.train.Int64List(value=[label_int])) })) # 寫入文件中，要先把協議序列化值之后才能存儲 writer.write(example.SerializeToString()) writer.close() return None def read_tfrecords(self): """ 從tfrecords文件讀取內容 :return: image_batch, label_batch """ # 構造文件隊列 file_q = tf.train.string_input_producer([flags.data_tfrecords]) # 構造閱讀器，讀取數據 reader = tf.TFRecordReader() # 一次只讀取一個樣本 key, value = reader.read(file_q) # 解析內容 解析example協議 feature = tf.parse_single_example(value, features={ "image":tf.FixedLenFeature([], tf.string), "label":tf.FixedLenFeature([], tf.int64) }) # 解碼 字符串需要解碼， 整形不用 image = tf.decode_raw(feature["image"], tf.uint8) # 設置圖片的形狀，以便批處理 image_reshape = tf.reshape(image, [self.height, self.width]) label = tf.cast(feature["label"], tf.int32) # 批處理 image_batch, label_batch = tf.train.batch([image_reshape, label],batch_size=10 ,num_threads=1, capacity=10) return image_batch, label_batch if __name__ == "__main__": filename = os.listdir(flags.data_home) filelist = [os.path.join(flags.data_home, file) for file in filename if file[-3:] == "bin"] cif = cifarread(filelist) # 讀取二進制文件 image_batch, label_batch = cif.read_decode() # 讀取tfrecords文件 # cif.read_tfrecords() with tf.Session() as sess: # 構建線程協調器 coord = tf.train.Coordinator() # 開啟線程 threads = tf.train.start_queue_runners(sess, coord=coord) # 執行任務 print(sess.run([image_batch, label_batch])) # 存儲tfrecords文件 # cif.write2tfrecords(image_batch, label_batch) # 回收線程 coord.request_stop() coord.join(threads)