机器学习Tensorflow基本操作：线程队列图像
一、線程和隊列
在使用TensorFlow進(jìn)行異步計算時，隊列是一種強(qiáng)大的機(jī)制。
為了感受一下隊列，讓我們來看一個簡單的例子。我們先創(chuàng)建一個“先入先出”的隊列（FIFOQueue），并將其內(nèi)部所有元素初始化為零。然后，我們構(gòu)建一個TensorFlow圖，它從隊列前端取走一個元素，加上1之后，放回隊列的后端。慢慢地，隊列的元素的值就會增加。
TensorFlow提供了兩個類來幫助多線程的實(shí)現(xiàn)：tf.Coordinator和 tf.QueueRunner。Coordinator類可以用來同時停止多個工作線程并且向那個在等待所有工作線程終止的程序報告異常，QueueRunner類用來協(xié)調(diào)多個工作線程同時將多個張量推入同一個隊列中。
隊列概述
隊列，如FIFOQueue和RandomShuffleQueue，在TensorFlow的張量異步計算時都非常重要。
例如，一個典型的輸入結(jié)構(gòu)：是使用一個RandomShuffleQueue來作為模型訓(xùn)練的輸入：
- 多個線程準(zhǔn)備訓(xùn)練樣本，并且把這些樣本推入隊列。
- 一個訓(xùn)練線程執(zhí)行一個訓(xùn)練操作
同步執(zhí)行隊列
# 創(chuàng)建一個隊列 Q = tf.FIFOQueue(3, dtypes=tf.float32)# 數(shù)據(jù)進(jìn)隊列 init = Q.enqueue_many(([0.1, 0.2, 0.3],))# 定義操作,op，出隊列，+1，進(jìn)隊列,注意返回的都是op out_q = Q.dequeue() data = out_q + 1 en_q = Q.enqueue(data)with tf.Session() as sess:# 初始化隊列，是數(shù)據(jù)進(jìn)入sess.run(init)# 執(zhí)行兩次入隊加1for i in range(2):sess.run(en_q)# 循環(huán)取隊列for i in range(3):print(sess.run(Q.dequeue()))tf.QueueRunner
QueueRunner類會創(chuàng)建一組線程， 這些線程可以重復(fù)的執(zhí)行Enquene操作， 他們使用同一個Coordinator來處理線程同步終止。此外，一個QueueRunner會運(yùn)行一個closer thread，當(dāng)Coordinator收到異常報告時，這個closer thread會自動關(guān)閉隊列。
您可以使用一個queue runner，來實(shí)現(xiàn)上述結(jié)構(gòu)。 首先建立一個TensorFlow圖表，這個圖表使用隊列來輸入樣本。增加處理樣本并將樣本推入隊列中的操作。增加training操作來移除隊列中的樣本。
tf.Coordinator
Coordinator類用來幫助多個線程協(xié)同工作，多個線程同步終止。 其主要方法有：
- should_stop():如果線程應(yīng)該停止則返回True。
- request_stop(): 請求該線程停止。
- join():等待被指定的線程終止。
首先創(chuàng)建一個Coordinator對象，然后建立一些使用Coordinator對象的線程。這些線程通常一直循環(huán)運(yùn)行，一直到should_stop()返回True時停止。 任何線程都可以決定計算什么時候應(yīng)該停止。它只需要調(diào)用request_stop()，同時其他線程的should_stop()將會返回True，然后都停下來。
異步執(zhí)行隊列：
#主線程，不斷的去取數(shù)據(jù)，開啟其它線程來進(jìn)行增加計數(shù)，入隊 #主線程結(jié)束了，隊列線程沒有結(jié)束，就會拋出異常 #主線程沒有結(jié)束，需要將隊列線程關(guān)閉，防止主線程等待Q = tf.FIFOQueue(1000,dtypes=tf.float32)# 定義操作 var = tf.Variable(0.0) increment_op = tf.assign_add(var,tf.constant(1.0)) en_op = Q.enqueue(increment_op)# 創(chuàng)建一個隊列管理器，指定線程數(shù)，執(zhí)行隊列的操作 qr = tf.train.QueueRunner(Q,enqueue_ops=[increment_op,en_op]*3)with tf.Session() as sess:tf.global_variables_initializer().run()# 生成一個線程協(xié)調(diào)器coord = tf.train.Coordinator()# 啟動線程執(zhí)行操作threads_list = qr.create_threads(sess,coord=coord,start=True)print(len(threads_list),"----------")# 主線程去取數(shù)據(jù)for i in range(20):print(sess.run(Q.dequeue()))# 請求其它線程終止coord.request_stop()# 關(guān)閉線程coord.join(threads_list)二、讀取數(shù)據(jù)
小數(shù)量數(shù)據(jù)讀取
這僅用于可以完全加載到存儲器中的小的數(shù)據(jù)集有兩種方法：
- 存儲在常數(shù)中。
- 存儲在變量中，初始化后，永遠(yuǎn)不要改變它的值。
使用常數(shù)更簡單一些，但是會使用更多的內(nèi)存，因?yàn)槌?shù)會內(nèi)聯(lián)的存儲在數(shù)據(jù)流圖數(shù)據(jù)結(jié)構(gòu)中，這個結(jié)構(gòu)體可能會被復(fù)制幾次。
training_data = ... training_labels = ... with tf.Session():input_data = tf.constant(training_data)input_labels = tf.constant(training_labels)要改為使用變量的方式，您就需要在數(shù)據(jù)流圖建立后初始化這個變量。
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_initalizer, trainable=False, collections=[])input_labels = tf.Variable(label_initalizer, trainable=False, collections=[])...sess.run(input_data.initializer,feed_dict={data_initializer: training_data})sess.run(input_labels.initializer,feed_dict={label_initializer: training_lables})設(shè)定trainable=False可以防止該變量被數(shù)據(jù)流圖的GraphKeys.TRAINABLE_VARIABLES收集，這樣我們就不會在訓(xùn)練的時候嘗試更新它的值;設(shè)定collections=[]可以防止GraphKeys.VARIABLES收集后做為保存和恢復(fù)的中斷點(diǎn)。設(shè)定這些標(biāo)志，是為了減少額外的開銷
文件讀取
先看下文件讀取以及讀取數(shù)據(jù)處理成張量結(jié)果的過程：
一般數(shù)據(jù)文件格式有文本、excel和圖片數(shù)據(jù)。那么TensorFlow都有對應(yīng)的解析函數(shù)，除了這幾種。還有TensorFlow指定的文件格式。
標(biāo)準(zhǔn)TensorFlow格式
TensorFlow還提供了一種內(nèi)置文件格式TFRecord，二進(jìn)制數(shù)據(jù)和訓(xùn)練類別標(biāo)簽數(shù)據(jù)存儲在同一文件。模型訓(xùn)練前圖像等文本信息轉(zhuǎn)換為TFRecord格式。TFRecord文件是protobuf格式。數(shù)據(jù)不壓縮，可快速加載到內(nèi)存。TFRecords文件包含 tf.train.Example protobuf，需要將Example填充到協(xié)議緩沖區(qū)，將協(xié)議緩沖區(qū)序列化為字符串，然后使用該文件將該字符串寫入TFRecords文件。在圖像操作我們會介紹整個過程以及詳細(xì)參數(shù)。
數(shù)據(jù)讀取實(shí)現(xiàn)
文件隊列生成函數(shù)
- tf.train.string_input_producer(string_tensor, num_epochs=None, shuffle=True, seed=None, capacity=32, name=None)
產(chǎn)生指定文件張量
文件閱讀器類
- class tf.TextLineReader
閱讀文本文件逗號分隔值（CSV）格式
- tf.FixedLengthRecordReader
要讀取每個記錄是固定數(shù)量字節(jié)的二進(jìn)制文件
- tf.TFRecordReader
讀取TfRecords文件
解碼
由于從文件中讀取的是字符串，需要函數(shù)去解析這些字符串到張量
tf.decode_csv（records，record_defaults,field_delim = None，name = None）將CSV轉(zhuǎn)換為張量，與tf.TextLineReader搭配使用
tf.decode_raw（bytes，out_type,little_endian = None，name = None） 將字節(jié)轉(zhuǎn)換為一個數(shù)字向量表示，字節(jié)為一字符串類型的張量,與函數(shù)tf.FixedLengthRecordReader搭配使用
生成文件隊列
將文件名列表交給tf.train.string_input_producer函數(shù)。string_input_producer來生成一個先入先出的隊列，文件閱讀器會需要它們來取數(shù)據(jù)。string_input_producer提供的可配置參數(shù)來設(shè)置文件名亂序和最大的訓(xùn)練迭代數(shù)，QueueRunner會為每次迭代（epoch）將所有的文件名加入文件名隊列中，如果shuffle=True的話，會對文件名進(jìn)行亂序處理。一過程是比較均勻的，因此它可以產(chǎn)生均衡的文件名隊列。
這個QueueRunner工作線程是獨(dú)立于文件閱讀器的線程，因此亂序和將文件名推入到文件名隊列這些過程不會阻塞文件閱讀器運(yùn)行。根據(jù)你的文件格式，選擇對應(yīng)的文件閱讀器，然后將文件名隊列提供給閱讀器的 read 方法。閱讀器的read方法會輸出一個鍵來表征輸入的文件和其中紀(jì)錄（對于調(diào)試非常有用），同時得到一個字符串標(biāo)量，這個字符串標(biāo)量可以被一個或多個解析器，或者轉(zhuǎn)換操作將其解碼為張量并且構(gòu)造成為樣本。
# 讀取CSV格式文件 # 1、構(gòu)建文件隊列# 2、構(gòu)建讀取器，讀取內(nèi)容# 3、解碼內(nèi)容# 4、現(xiàn)讀取一個內(nèi)容，如果有需要，就批處理內(nèi)容 import tensorflow as tf import os def readcsv_decode(filelist):"""讀取并解析文件內(nèi)容:param filelist: 文件列表:return: None"""# 把文件目錄和文件名合并flist = [os.path.join("./csvdata/",file) for file in filelist]# 構(gòu)建文件隊列file_queue = tf.train.string_input_producer(flist,shuffle=False)# 構(gòu)建閱讀器，讀取文件內(nèi)容reader = tf.TextLineReader()key,value = reader.read(file_queue)record_defaults = [["null"],["null"]] # [[0],[0],[0],[0]]# 解碼內(nèi)容，按行解析，返回的是每行的列數(shù)據(jù)example,label = tf.decode_csv(value,record_defaults=record_defaults)# 通過tf.train.batch來批處理數(shù)據(jù)example_batch,label_batch = tf.train.batch([example,label],batch_size=9,num_threads=1,capacity=9)with tf.Session() as sess:# 線程協(xié)調(diào)員coord = tf.train.Coordinator()# 啟動工作線程threads = tf.train.start_queue_runners(sess,coord=coord)# 這種方法不可取# for i in range(9):# print(sess.run([example,label]))# 打印批處理的數(shù)據(jù)print(sess.run([example_batch,label_batch]))coord.request_stop()coord.join(threads)return Noneif __name__=="__main__":filename_list = os.listdir("./csvdata")readcsv_decode(filename_list)每次read的執(zhí)行都會從文件中讀取一行內(nèi)容，注意，(這與后面的圖片和TfRecords讀取不一樣)，decode_csv操作會解析這一行內(nèi)容并將其轉(zhuǎn)為張量列表。如果輸入的參數(shù)有缺失，record_default參數(shù)可以根據(jù)張量的類型來設(shè)置默認(rèn)值。在調(diào)用run或者eval去執(zhí)行read之前，你必須調(diào)用tf.train.start_queue_runners來將文件名填充到隊列。否則read操作會被阻塞到文件名隊列中有值為止。
三、圖像操作
圖像基本概念
在圖像數(shù)字化表示當(dāng)中，分為黑白和彩色兩種。在數(shù)字化表示圖片的時候，有三個因素。分別是圖片的長、圖片的寬、圖片的顏色通道數(shù)。那么黑白圖片的顏色通道數(shù)為1，它只需要一個數(shù)字就可以表示一個像素位；而彩色照片就不一樣了，它有三個顏色通道，分別為RGB，通過三個數(shù)字表示一個像素位。TensorFlow支持JPG、PNG圖像格式，RGB、RGBA顏色空間。圖像用與圖像尺寸相同(heightwidthchnanel)張量表示。圖像所有像素存在磁盤文件，需要被加載到內(nèi)存。
圖像大小壓縮
大尺寸圖像輸入占用大量系統(tǒng)內(nèi)存。訓(xùn)練CNN需要大量時間，加載大文件增加更多訓(xùn)練時間，也難存放多數(shù)系統(tǒng)GPU顯存。大尺寸圖像大量無關(guān)本征屬性信息，影響模型泛化能力。最好在預(yù)處理階段完成圖像操作，縮小、裁剪、縮放、灰度調(diào)整等。圖像加載后，翻轉(zhuǎn)、扭曲，使輸入網(wǎng)絡(luò)訓(xùn)練信息多樣化，緩解過擬合。Python圖像處理框架PIL、OpenCV。TensorFlow提供部分圖像處理方法。
- tf.image.resize_images 壓縮圖片導(dǎo)致定大小
圖像數(shù)據(jù)讀取實(shí)例
同樣圖像加載與二進(jìn)制文件相同。圖像需要解碼。輸入生成器(tf.train.string_input_producer)找到所需文件，加載到隊列。tf.WholeFileReader 加載完整圖像文件到內(nèi)存，WholeFileReader.read 讀取圖像，tf.image.decode_jpeg 解碼JPEG格式圖像。圖像是三階張量。RGB值是一階張量。加載圖像格式為[batch_size,image_height,image_width,channels]。批數(shù)據(jù)圖像過大過多，占用內(nèi)存過高，系統(tǒng)會停止響應(yīng)。直接加載TFRecord文件，可以節(jié)省訓(xùn)練時間。支持寫入多個樣本。
讀取圖片數(shù)據(jù)到Tensor
管道讀端多文件內(nèi)容處理
但是會發(fā)現(xiàn)read只返回一個圖片的值。所以我們在之前處理文件的整個流程中，后面的內(nèi)容隊列的出隊列需要用特定函數(shù)去獲取。
- tf.train.batch 讀取指定大小（個數(shù)）的張量
- tf.train.shuffle_batch 亂序讀取指定大小（個數(shù)）的張量
讀取TfRecords文件數(shù)據(jù)
#CIFAR-10的數(shù)據(jù)讀取以及轉(zhuǎn)換成TFRecordsg格式#1、數(shù)據(jù)的讀取FLAGS = tf.app.flags.FLAGStf.app.flags.DEFINE_string("data_dir","./cifar10/cifar-10-batches-bin/","CIFAR數(shù)據(jù)目錄") tf.app.flags.DEFINE_integer("batch_size",50000,"樣本個數(shù)") tf.app.flags.DEFINE_string("records_file","./cifar10/cifar.tfrecords","tfrecords文件位置")class CifarRead(object):def __init__(self,filename):self.filelist = filename# 定義圖片的長、寬、深度，標(biāo)簽字節(jié)，圖像字節(jié)，總字節(jié)數(shù)self.height = 32self.width = 32self.depth = 3self.label_bytes = 1self.image_bytes = self.height*self.width*self.depthself.bytes = self.label_bytes + self.image_bytesdef readcifar_decode(self):"""讀取數(shù)據(jù)，進(jìn)行轉(zhuǎn)換:return: 批處理的圖片和標(biāo)簽"""# 1、構(gòu)造文件隊列file_queue = tf.train.string_input_producer(self.filelist)# 2、構(gòu)造讀取器，讀取內(nèi)容reader = tf.FixedLengthRecordReader(self.bytes)key,value = reader.read(file_queue)# 3、文件內(nèi)容解碼image_label = tf.decode_raw(value,tf.uint8)# 分割標(biāo)簽與圖像張量，轉(zhuǎn)換成相應(yīng)的格式label = tf.cast(tf.slice(image_label,[0],[self.label_bytes]),tf.int32)image = tf.slice(image_label,[self.label_bytes],[self.image_bytes])print(image)# 給image設(shè)置形狀，防止批處理出錯image_tensor = tf.reshape(image,[self.height,self.width,self.depth])print(image_tensor.eval())# depth_major = tf.reshape(image, [self.depth,self.height, self.width])# image_tensor = tf.transpose(depth_major, [1, 2, 0])# 4、處理流程image_batch,label_batch = tf.train.batch([image_tensor,label],batch_size=10,num_threads=1,capacity=10)return image_batch,label_batchdef convert_to_tfrecords(self,image_batch,label_batch):"""轉(zhuǎn)換成TFRecords文件:param image_batch: 圖片數(shù)據(jù)Tensor:param label_batch: 標(biāo)簽數(shù)據(jù)Tensor:param sess: 會話:return: None"""# 創(chuàng)建一個TFRecord存儲器writer = tf.python_io.TFRecordWriter(FLAGS.records_file)# 構(gòu)造每個樣本的Examplefor i in range(10):print("---------")image = image_batch[i]# 將單個圖片張量轉(zhuǎn)換為字符串，以可以存進(jìn)二進(jìn)制文件image_string = image.eval().tostring()# 使用eval需要注意的是，必須存在會話上下文環(huán)境label = int(label_batch[i].eval()[0])# 構(gòu)造協(xié)議塊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]))}))# 寫進(jìn)文件writer.write(example.SerializeToString())writer.close()return Nonedef read_from_tfrecords(self):"""讀取tfrecords:return: None"""file_queue = tf.train.string_input_producer(["./cifar10/cifar.tfrecords"])reader = tf.TFRecordReader()key, value = reader.read(file_queue)features = tf.parse_single_example(value, features={"image":tf.FixedLenFeature([], tf.string),"label":tf.FixedLenFeature([], tf.int64),})image = tf.decode_raw(features["image"], tf.uint8)# 設(shè)置靜態(tài)形狀，可用于轉(zhuǎn)換動態(tài)形狀image.set_shape([self.image_bytes])print(image)image_tensor = tf.reshape(image,[self.height,self.width,self.depth])print(image_tensor)label = tf.cast(features["label"], tf.int32)print(label)image_batch, label_batch = tf.train.batch([image_tensor, label],batch_size=10,num_threads=1,capacity=10)print(image_batch)print(label_batch)with tf.Session() as sess:coord = tf.train.Coordinator()threads = tf.train.start_queue_runners(sess=sess,coord=coord)print(sess.run([image_batch, label_batch]))coord.request_stop()coord.join(threads)return Noneif __name__=="__main__":# 構(gòu)造文件名字的列表filename = os.listdir(FLAGS.data_dir)file_list = [os.path.join(FLAGS.data_dir, file) for file in filename if file[-3:] == "bin"]cfar = CifarRead(file_list)# image_batch,label_batch = cfar.readcifar_decode()cfar.read_from_tfrecords()with tf.Session() as sess:# 構(gòu)建線程協(xié)調(diào)器coord = tf.train.Coordinator()# 開啟線程threads = tf.train.start_queue_runners(sess=sess, coord=coord)# print(sess.run(image_batch))# 存進(jìn)文件# cfar.convert_to_tfrecords(image_batch, label_batch)coord.request_stop()coord.join(threads)總結(jié)
以上是生活随笔為你收集整理的机器学习Tensorflow基本操作：线程队列图像的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 微服务 注册中心的作用_102，谈谈微服
- 下一篇： VUE：解决判断网页端与手机端情况下，横