Tensorflow |（6）Tensorflow的IO操作

Tensorflow |（1）初識(shí)Tensorflow

Tensorflow |（2）張量的階和數(shù)據(jù)類型及張量操作

Tensorflow |（3）變量的的創(chuàng)建、初始化、保存和加載

Tensorflow |（4）名稱域、圖 和會(huì)話

Tensorflow |（5）模型保存與恢復(fù)、自定義命令行參數(shù)

Tensorflow |（6）Tensorflow的IO操作

讀取數(shù)據(jù)

小數(shù)量數(shù)據(jù)讀取

這僅用于可以完全加載到存儲(chǔ)器中的小的數(shù)據(jù)集有兩種方法：

存儲(chǔ)在常數(shù)中。

存儲(chǔ)在變量中，初始化后，永遠(yuǎn)不要改變它的值。

使用常數(shù)更簡單一些，但是會(huì)使用更多的內(nèi)存，因?yàn)槌?shù)會(huì)內(nèi)聯(lián)的存儲(chǔ)在數(shù)據(jù)流圖數(shù)據(jù)結(jié)構(gòu)中，這個(gè)結(jié)構(gòu)體可能會(huì)被復(fù)制幾次。

training_data = ...

training_labels = ...

with tf.Session():

input_data = tf.constant(training_data)

input_labels = tf.constant(training_labels)

要改為使用變量的方式，您就需要在數(shù)據(jù)流圖建立后初始化這個(gè)變量。

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收集，這樣我們就不會(huì)在訓(xùn)練的時(shí)候嘗試更新它的值;設(shè)定collections=[]可以防止GraphKeys.VARIABLES收集后做為保存和恢復(fù)的中斷點(diǎn)。設(shè)定這些標(biāo)志，是為了減少額外的開銷

文件讀取

先看下文件讀取以及讀取數(shù)據(jù)處理成張量結(jié)果的過程：

一般數(shù)據(jù)文件格式有文本、excel和圖片數(shù)據(jù)。那么TensorFlow都有對(duì)應(yīng)的解析函數(shù)，除了這幾種。還有TensorFlow指定的文件格式。

標(biāo)準(zhǔn)TensorFlow格式

TensorFlow還提供了一種內(nèi)置文件格式TFRecord，二進(jìn)制數(shù)據(jù)和訓(xùn)練類別標(biāo)簽數(shù)據(jù)存儲(chǔ)在同一文件。模型訓(xùn)練前圖像等文本信息轉(zhuǎn)換為TFRecord格式。TFRecord文件是protobuf格式。數(shù)據(jù)不壓縮，可快速加載到內(nèi)存。TFRecords文件包含 tf.train.Example protobuf，需要將Example填充到協(xié)議緩沖區(qū)，將協(xié)議緩沖區(qū)序列化為字符串，然后使用該文件將該字符串寫入TFRecords文件。在圖像操作我們會(huì)介紹整個(gè)過程以及詳細(xì)參數(shù)。

數(shù)據(jù)讀取實(shí)現(xiàn)

文件隊(duì)列生成函數(shù)

tf.train.string_input_producer(string_tensor, num_epochs=None, shuffle=True, seed=None, capacity=32, name=None)

產(chǎn)生指定文件張量

文件閱讀器類

class tf.TextLineReader

閱讀文本文件逗號(hào)分隔值（CSV）格式

tf.FixedLengthRecordReader

要讀取每個(gè)記錄是固定數(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)換為一個(gè)數(shù)字向量表示，字節(jié)為一字符串類型的張量,與函數(shù)tf.FixedLengthRecordReader搭配使用

生成文件隊(duì)列

將文件名列表交給tf.train.string_input_producer函數(shù)。string_input_producer來生成一個(gè)先入先出的隊(duì)列，文件閱讀器會(huì)需要它們來取數(shù)據(jù)。string_input_producer提供的可配置參數(shù)來設(shè)置文件名亂序和最大的訓(xùn)練迭代數(shù)，QueueRunner會(huì)為每次迭代（epoch）將所有的文件名加入文件名隊(duì)列中，如果shuffle=True的話，會(huì)對(duì)文件名進(jìn)行亂序處理。一過程是比較均勻的，因此它可以產(chǎn)生均衡的文件名隊(duì)列。

這個(gè)QueueRunner工作線程是獨(dú)立于文件閱讀器的線程，因此亂序和將文件名推入到文件名隊(duì)列這些過程不會(huì)阻塞文件閱讀器運(yùn)行。根據(jù)你的文件格式，選擇對(duì)應(yīng)的文件閱讀器，然后將文件名隊(duì)列提供給閱讀器的?read?方法。閱讀器的read方法會(huì)輸出一個(gè)鍵來表征輸入的文件和其中紀(jì)錄（對(duì)于調(diào)試非常有用），同時(shí)得到一個(gè)字符串標(biāo)量，這個(gè)字符串標(biāo)量可以被一個(gè)或多個(gè)解析器，或者轉(zhuǎn)換操作將其解碼為張量并且構(gòu)造成為樣本。

# 讀取CSV格式文件

# 1、構(gòu)建文件隊(duì)列

# 2、構(gòu)建讀取器，讀取內(nèi)容

# 3、解碼內(nèi)容

# 4、現(xiàn)讀取一個(gè)內(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)建文件隊(duì)列

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()

# 啟動(dòng)工作線程

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 None

if __name__=="__main__":

filename_list = os.listdir("./csvdata")

readcsv_decode(filename_list)

每次read的執(zhí)行都會(huì)從文件中讀取一行內(nèi)容，注意，(這與后面的圖片和TfRecords讀取不一樣)，decode_csv操作會(huì)解析這一行內(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來將文件名填充到隊(duì)列。否則read操作會(huì)被阻塞到文件名隊(duì)列中有值為止。

線程和隊(duì)列

在使用TensorFlow進(jìn)行異步計(jì)算時(shí)，隊(duì)列是一種強(qiáng)大的機(jī)制。

為了感受一下隊(duì)列，讓我們來看一個(gè)簡單的例子。我們先創(chuàng)建一個(gè)“先入先出”的隊(duì)列（FIFOQueue），并將其內(nèi)部所有元素初始化為零。然后，我們構(gòu)建一個(gè)TensorFlow圖，它從隊(duì)列前端取走一個(gè)元素，加上1之后，放回隊(duì)列的后端。慢慢地，隊(duì)列的元素的值就會(huì)增加。

TensorFlow提供了兩個(gè)類來幫助多線程的實(shí)現(xiàn)：tf.Coordinator和 tf.QueueRunner。Coordinator類可以用來同時(shí)停止多個(gè)工作線程并且向那個(gè)在等待所有工作線程終止的程序報(bào)告異常，QueueRunner類用來協(xié)調(diào)多個(gè)工作線程同時(shí)將多個(gè)張量推入同一個(gè)隊(duì)列中。

隊(duì)列概述

隊(duì)列，如FIFOQueue和RandomShuffleQueue，在TensorFlow的張量異步計(jì)算時(shí)都非常重要。

例如，一個(gè)典型的輸入結(jié)構(gòu)：是使用一個(gè)RandomShuffleQueue來作為模型訓(xùn)練的輸入：

多個(gè)線程準(zhǔn)備訓(xùn)練樣本，并且把這些樣本推入隊(duì)列。

一個(gè)訓(xùn)練線程執(zhí)行一個(gè)訓(xùn)練操作

同步執(zhí)行隊(duì)列

# 創(chuàng)建一個(gè)隊(duì)列

Q = tf.FIFOQueue(3, dtypes=tf.float32)

# 數(shù)據(jù)進(jìn)隊(duì)列

init = Q.enqueue_many(([0.1, 0.2, 0.3],))

# 定義操作,op，出隊(duì)列，+1，進(jìn)隊(duì)列,注意返回的都是op

out_q = Q.dequeue()

data = out_q + 1

en_q = Q.enqueue(data)

with tf.Session() as sess:

# 初始化隊(duì)列，是數(shù)據(jù)進(jìn)入

sess.run(init)

# 執(zhí)行兩次入隊(duì)加1

for i in range(2):

sess.run(en_q)

# 循環(huán)取隊(duì)列

for i in range(3):

print(sess.run(Q.dequeue()))

tf.QueueRunner

QueueRunner類會(huì)創(chuàng)建一組線程， 這些線程可以重復(fù)的執(zhí)行Enquene操作， 他們使用同一個(gè)Coordinator來處理線程同步終止。此外，一個(gè)QueueRunner會(huì)運(yùn)行一個(gè)closer thread，當(dāng)Coordinator收到異常報(bào)告時(shí)，這個(gè)closer thread會(huì)自動(dòng)關(guān)閉隊(duì)列。

您可以使用一個(gè)queue runner，來實(shí)現(xiàn)上述結(jié)構(gòu)。 首先建立一個(gè)TensorFlow圖表，這個(gè)圖表使用隊(duì)列來輸入樣本。增加處理樣本并將樣本推入隊(duì)列中的操作。增加training操作來移除隊(duì)列中的樣本。

tf.Coordinator

Coordinator類用來幫助多個(gè)線程協(xié)同工作，多個(gè)線程同步終止。 其主要方法有：

should_stop():如果線程應(yīng)該停止則返回True。

request_stop(): 請(qǐng)求該線程停止。

join():等待被指定的線程終止。

首先創(chuàng)建一個(gè)Coordinator對(duì)象，然后建立一些使用Coordinator對(duì)象的線程。這些線程通常一直循環(huán)運(yùn)行，一直到should_stop()返回True時(shí)停止。 任何線程都可以決定計(jì)算什么時(shí)候應(yīng)該停止。它只需要調(diào)用request_stop()，同時(shí)其他線程的should_stop()將會(huì)返回True，然后都停下來。

異步執(zhí)行隊(duì)列：

#主線程，不斷的去取數(shù)據(jù)，開啟其它線程來進(jìn)行增加計(jì)數(shù)，入隊(duì)

#主線程結(jié)束了，隊(duì)列線程沒有結(jié)束，就會(huì)拋出異常

#主線程沒有結(jié)束，需要將隊(duì)列線程關(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)建一個(gè)隊(duì)列管理器，指定線程數(shù)，執(zhí)行隊(duì)列的操作

qr = tf.train.QueueRunner(Q,enqueue_ops=[increment_op,en_op]*3)

with tf.Session() as sess:

tf.global_variables_initializer().run()

# 生成一個(gè)線程協(xié)調(diào)器

coord = tf.train.Coordinator()

# 啟動(dòng)線程執(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()))

# 請(qǐng)求其它線程終止

coord.request_stop()

# 關(guān)閉線程

coord.join(threads_list)

圖像操作

圖像基本概念

在圖像數(shù)字化表示當(dāng)中，分為黑白和彩色兩種。在數(shù)字化表示圖片的時(shí)候，有三個(gè)因素。分別是圖片的長、圖片的寬、圖片的顏色通道數(shù)。那么黑白圖片的顏色通道數(shù)為1，它只需要一個(gè)數(shù)字就可以表示一個(gè)像素位；而彩色照片就不一樣了，它有三個(gè)顏色通道，分別為RGB，通過三個(gè)數(shù)字表示一個(gè)像素位。TensorFlow支持JPG、PNG圖像格式，RGB、RGBA顏色空間。圖像用與圖像尺寸相同(heightwidthchnanel)張量表示。圖像所有像素存在磁盤文件，需要被加載到內(nèi)存。

圖像大小壓縮

大尺寸圖像輸入占用大量系統(tǒng)內(nèi)存。訓(xùn)練CNN需要大量時(shí)間，加載大文件增加更多訓(xùn)練時(shí)間，也難存放多數(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)找到所需文件，加載到隊(duì)列。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)會(huì)停止響應(yīng)。直接加載TFRecord文件，可以節(jié)省訓(xùn)練時(shí)間。支持寫入多個(gè)樣本。

讀取圖片數(shù)據(jù)到Tensor

管道讀端多文件內(nèi)容處理

但是會(huì)發(fā)現(xiàn)read只返回一個(gè)圖片的值。所以我們?cè)谥疤幚砦募恼麄€(gè)流程中，后面的內(nèi)容隊(duì)列的出隊(duì)列需要用特定函數(shù)去獲取。

tf.train.batch?讀取指定大小（個(gè)數(shù)）的張量

tf.train.shuffle_batch?亂序讀取指定大小（個(gè)數(shù)）的張量

def readpic_decode(file_list):

"""

批量讀取圖片并轉(zhuǎn)換成張量格式

:param file_list: 文件名目錄列表

:return: None

"""

# 構(gòu)造文件隊(duì)列

file_queue = tf.train.string_input_producer(file_list)

# 圖片閱讀器和讀取數(shù)據(jù)

reader = tf.WholeFileReader()

key,value = reader.read(file_queue)

# 解碼成張量形式

image_first = tf.image.decode_jpeg(value)

print(image_first)

# 縮小圖片到指定長寬，不用指定通道數(shù)

image = tf.image.resize_images(image_first,[256,256])

# 設(shè)置圖片的靜態(tài)形狀

image.set_shape([256,256,3])

print(image)

# 批處理圖片數(shù)據(jù)，tensors是需要具體的形狀大小

image_batch = tf.train.batch([image],batch_size=100,num_threads=1,capacity=100)

tf.summary.image("pic",image_batch)

with tf.Session() as sess:

merged = tf.summary.merge_all()

filewriter = tf.summary.FileWriter("/tmp/summary/dog/",graph=sess.graph)

# 線程協(xié)調(diào)器

coord = tf.train.Coordinator()

# 開啟線程

threads = tf.train.start_queue_runners(sess=sess,coord=coord)

print(sess.run(image_batch))

summary = sess.run(merged)

filewriter.add_summary(summary)

# 等待線程回收

coord.request_stop()

coord.join(threads)

return None

if __name__=="__main__":

# 獲取文件列表

filename = os.listdir("./dog/")

# 組合文件目錄和文件名

file_list = [os.path.join("./dog/",file) for file in filename]

# 調(diào)用讀取函數(shù)

readpic_decode(file_list)

讀取TfRecords文件數(shù)據(jù)

#CIFAR-10的數(shù)據(jù)讀取以及轉(zhuǎn)換成TFRecordsg格式

#1、數(shù)據(jù)的讀取

FLAGS = tf.app.flags.FLAGS

tf.app.flags.DEFINE_string("data_dir","./cifar10/cifar-10-batches-bin/","CIFAR數(shù)據(jù)目錄")

tf.app.flags.DEFINE_integer("batch_size",50000,"樣本個(gè)數(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 = 32

self.width = 32

self.depth = 3

self.label_bytes = 1

self.image_bytes = self.height*self.width*self.depth

self.bytes = self.label_bytes + self.image_bytes

def readcifar_decode(self):

"""

讀取數(shù)據(jù)，進(jìn)行轉(zhuǎn)換

:return: 批處理的圖片和標(biāo)簽

"""

# 1、構(gòu)造文件隊(duì)列

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è)置形狀，防止批處理出錯(cuò)

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_batch

def 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: 會(huì)話

:return: None

"""

# 創(chuàng)建一個(gè)TFRecord存儲(chǔ)器

writer = tf.python_io.TFRecordWriter(FLAGS.records_file)

# 構(gòu)造每個(gè)樣本的Example

for i in range(10):

print("---------")

image = image_batch[i]

# 將單個(gè)圖片張量轉(zhuǎn)換為字符串，以可以存進(jìn)二進(jìn)制文件

image_string = image.eval().tostring()

# 使用eval需要注意的是，必須存在會(huì)話上下文環(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 None

def 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)換動(dòng)態(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 None

if __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)

TensorFlow 任務(wù)調(diào)度

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實(shí)的內(nèi)容，請(qǐng)聯(lián)系我們jiasou666@gmail.com 處理，核實(shí)后本網(wǎng)站將在24小時(shí)內(nèi)刪除侵權(quán)內(nèi)容。

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實(shí)的內(nèi)容，請(qǐng)聯(lián)系我們jiasou666@gmail.com 處理，核實(shí)后本網(wǎng)站將在24小時(shí)內(nèi)刪除侵權(quán)內(nèi)容。