机器学习Tensorflow基本操作：线程队列图像
一、線程和隊列
在使用TensorFlow進行異步計算時，隊列是一種強大的機制。
為了感受一下隊列，讓我們來看一個簡單的例子。我們先創(chuàng)建一個“先入先出”的隊列（FIFOQueue），并將其內(nèi)部所有元素初始化為零。然后，我們構建一個TensorFlow圖，它從隊列前端取走一個元素，加上1之后，放回隊列的后端。慢慢地，隊列的元素的值就會增加。
TensorFlow提供了兩個類來幫助多線程的實現(xiàn)：tf.Coordinator和 tf.QueueRunner。Coordinator類可以用來同時停止多個工作線程并且向那個在等待所有工作線程終止的程序報告異常，QueueRunner類用來協(xié)調多個工作線程同時將多個張量推入同一個隊列中。
隊列概述
隊列，如FIFOQueue和RandomShuffleQueue，在TensorFlow的張量異步計算時都非常重要。
例如，一個典型的輸入結構：是使用一個RandomShuffleQueue來作為模型訓練的輸入：
- 多個線程準備訓練樣本，并且把這些樣本推入隊列。
- 一個訓練線程執(zhí)行一個訓練操作
同步執(zhí)行隊列
# 創(chuàng)建一個隊列 Q = tf.FIFOQueue(3, dtypes=tf.float32)# 數(shù)據(jù)進隊列 init = Q.enqueue_many(([0.1, 0.2, 0.3],))# 定義操作,op，出隊列，+1，進隊列,注意返回的都是op out_q = Q.dequeue() data = out_q + 1 en_q = Q.enqueue(data)with tf.Session() as sess:# 初始化隊列，是數(shù)據(jù)進入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)建一組線程， 這些線程可以重復的執(zhí)行Enquene操作， 他們使用同一個Coordinator來處理線程同步終止。此外，一個QueueRunner會運行一個closer thread，當Coordinator收到異常報告時，這個closer thread會自動關閉隊列。
您可以使用一個queue runner，來實現(xiàn)上述結構。 首先建立一個TensorFlow圖表，這個圖表使用隊列來輸入樣本。增加處理樣本并將樣本推入隊列中的操作。增加training操作來移除隊列中的樣本。
tf.Coordinator
Coordinator類用來幫助多個線程協(xié)同工作，多個線程同步終止。 其主要方法有：
- should_stop():如果線程應該停止則返回True。
- request_stop(): 請求該線程停止。
- join():等待被指定的線程終止。
首先創(chuàng)建一個Coordinator對象，然后建立一些使用Coordinator對象的線程。這些線程通常一直循環(huán)運行，一直到should_stop()返回True時停止。 任何線程都可以決定計算什么時候應該停止。它只需要調用request_stop()，同時其他線程的should_stop()將會返回True，然后都停下來。
異步執(zhí)行隊列：
#主線程，不斷的去取數(shù)據(jù)，開啟其它線程來進行增加計數(shù)，入隊 #主線程結束了，隊列線程沒有結束，就會拋出異常 #主線程沒有結束，需要將隊列線程關閉，防止主線程等待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é)調器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()# 關閉線程coord.join(threads_list)二、讀取數(shù)據(jù)
小數(shù)量數(shù)據(jù)讀取
這僅用于可以完全加載到存儲器中的小的數(shù)據(jù)集有兩種方法：
- 存儲在常數(shù)中。
- 存儲在變量中，初始化后，永遠不要改變它的值。
使用常數(shù)更簡單一些，但是會使用更多的內(nèi)存，因為常數(shù)會內(nèi)聯(lián)的存儲在數(shù)據(jù)流圖數(shù)據(jù)結構中，這個結構體可能會被復制幾次。
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})設定trainable=False可以防止該變量被數(shù)據(jù)流圖的GraphKeys.TRAINABLE_VARIABLES收集，這樣我們就不會在訓練的時候嘗試更新它的值;設定collections=[]可以防止GraphKeys.VARIABLES收集后做為保存和恢復的中斷點。設定這些標志，是為了減少額外的開銷
文件讀取
先看下文件讀取以及讀取數(shù)據(jù)處理成張量結果的過程：
一般數(shù)據(jù)文件格式有文本、excel和圖片數(shù)據(jù)。那么TensorFlow都有對應的解析函數(shù)，除了這幾種。還有TensorFlow指定的文件格式。
標準TensorFlow格式
TensorFlow還提供了一種內(nèi)置文件格式TFRecord，二進制數(shù)據(jù)和訓練類別標簽數(shù)據(jù)存儲在同一文件。模型訓練前圖像等文本信息轉換為TFRecord格式。TFRecord文件是protobuf格式。數(shù)據(jù)不壓縮，可快速加載到內(nèi)存。TFRecords文件包含 tf.train.Example protobuf，需要將Example填充到協(xié)議緩沖區(qū)，將協(xié)議緩沖區(qū)序列化為字符串，然后使用該文件將該字符串寫入TFRecords文件。在圖像操作我們會介紹整個過程以及詳細參數(shù)。
數(shù)據(jù)讀取實現(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é)的二進制文件
- tf.TFRecordReader
讀取TfRecords文件
解碼
由于從文件中讀取的是字符串，需要函數(shù)去解析這些字符串到張量
tf.decode_csv（records，record_defaults,field_delim = None，name = None）將CSV轉換為張量，與tf.TextLineReader搭配使用
tf.decode_raw（bytes，out_type,little_endian = None，name = None） 將字節(jié)轉換為一個數(shù)字向量表示，字節(jié)為一字符串類型的張量,與函數(shù)tf.FixedLengthRecordReader搭配使用
生成文件隊列
將文件名列表交給tf.train.string_input_producer函數(shù)。string_input_producer來生成一個先入先出的隊列，文件閱讀器會需要它們來取數(shù)據(jù)。string_input_producer提供的可配置參數(shù)來設置文件名亂序和最大的訓練迭代數(shù)，QueueRunner會為每次迭代（epoch）將所有的文件名加入文件名隊列中，如果shuffle=True的話，會對文件名進行亂序處理。一過程是比較均勻的，因此它可以產(chǎn)生均衡的文件名隊列。
這個QueueRunner工作線程是獨立于文件閱讀器的線程，因此亂序和將文件名推入到文件名隊列這些過程不會阻塞文件閱讀器運行。根據(jù)你的文件格式，選擇對應的文件閱讀器，然后將文件名隊列提供給閱讀器的 read 方法。閱讀器的read方法會輸出一個鍵來表征輸入的文件和其中紀錄（對于調試非常有用），同時得到一個字符串標量，這個字符串標量可以被一個或多個解析器，或者轉換操作將其解碼為張量并且構造成為樣本。
# 讀取CSV格式文件 # 1、構建文件隊列# 2、構建讀取器，讀取內(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]# 構建文件隊列file_queue = tf.train.string_input_producer(flist,shuffle=False)# 構建閱讀器，讀取文件內(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é)調員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)容并將其轉為張量列表。如果輸入的參數(shù)有缺失，record_default參數(shù)可以根據(jù)張量的類型來設置默認值。在調用run或者eval去執(zhí)行read之前，你必須調用tf.train.start_queue_runners來將文件名填充到隊列。否則read操作會被阻塞到文件名隊列中有值為止。
三、圖像操作
圖像基本概念
在圖像數(shù)字化表示當中，分為黑白和彩色兩種。在數(shù)字化表示圖片的時候，有三個因素。分別是圖片的長、圖片的寬、圖片的顏色通道數(shù)。那么黑白圖片的顏色通道數(shù)為1，它只需要一個數(shù)字就可以表示一個像素位；而彩色照片就不一樣了，它有三個顏色通道，分別為RGB，通過三個數(shù)字表示一個像素位。TensorFlow支持JPG、PNG圖像格式，RGB、RGBA顏色空間。圖像用與圖像尺寸相同(heightwidthchnanel)張量表示。圖像所有像素存在磁盤文件，需要被加載到內(nèi)存。
圖像大小壓縮
大尺寸圖像輸入占用大量系統(tǒng)內(nèi)存。訓練CNN需要大量時間，加載大文件增加更多訓練時間，也難存放多數(shù)系統(tǒng)GPU顯存。大尺寸圖像大量無關本征屬性信息，影響模型泛化能力。最好在預處理階段完成圖像操作，縮小、裁剪、縮放、灰度調整等。圖像加載后，翻轉、扭曲，使輸入網(wǎng)絡訓練信息多樣化，緩解過擬合。Python圖像處理框架PIL、OpenCV。TensorFlow提供部分圖像處理方法。
- tf.image.resize_images 壓縮圖片導致定大小
圖像數(shù)據(jù)讀取實例
同樣圖像加載與二進制文件相同。圖像需要解碼。輸入生成器(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)會停止響應。直接加載TFRecord文件，可以節(jié)省訓練時間。支持寫入多個樣本。
讀取圖片數(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ù)讀取以及轉換成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# 定義圖片的長、寬、深度，標簽字節(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ù)，進行轉換:return: 批處理的圖片和標簽"""# 1、構造文件隊列file_queue = tf.train.string_input_producer(self.filelist)# 2、構造讀取器，讀取內(nèi)容reader = tf.FixedLengthRecordReader(self.bytes)key,value = reader.read(file_queue)# 3、文件內(nèi)容解碼image_label = tf.decode_raw(value,tf.uint8)# 分割標簽與圖像張量，轉換成相應的格式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設置形狀，防止批處理出錯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):"""轉換成TFRecords文件:param image_batch: 圖片數(shù)據(jù)Tensor:param label_batch: 標簽數(shù)據(jù)Tensor:param sess: 會話:return: None"""# 創(chuàng)建一個TFRecord存儲器writer = tf.python_io.TFRecordWriter(FLAGS.records_file)# 構造每個樣本的Examplefor i in range(10):print("---------")image = image_batch[i]# 將單個圖片張量轉換為字符串，以可以存進二進制文件image_string = image.eval().tostring()# 使用eval需要注意的是，必須存在會話上下文環(huán)境label = int(label_batch[i].eval()[0])# 構造協(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]))}))# 寫進文件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)# 設置靜態(tài)形狀，可用于轉換動態(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__":# 構造文件名字的列表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:# 構建線程協(xié)調器coord = tf.train.Coordinator()# 開啟線程threads = tf.train.start_queue_runners(sess=sess, coord=coord)# print(sess.run(image_batch))# 存進文件# cfar.convert_to_tfrecords(image_batch, label_batch)coord.request_stop()coord.join(threads)總結
以上是生活随笔為你收集整理的机器学习Tensorflow基本操作：线程队列图像的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 微服务 注册中心的作用_102，谈谈微服
- 下一篇： VUE：解决判断网页端与手机端情况下，横