檔案讀取流程

學習目標

• 目標
  • 說明TensorFlow檔案讀取的流程
• 應用
  • 無

有四種獲取資料到TensorFlow程式的方法：

1. tf.dataAPI：輕鬆構建複雜的輸入管道。（優選方法，在新版本當中）
2. QueueRunner：基於佇列的輸入管道從TensorFlow圖形開頭的檔案中讀取資料。
3. Feeding：執行每一步時，Python程式碼提供資料。
4. 預載入資料：TensorFlow圖中的常量或變數包含所有資料（對於小資料集）。

1、檔案讀取流程

• 第一階段將生成檔名來讀取它們並將它們排入檔名佇列。
• 第二階段對於檔名的佇列，進行出佇列例項，並且實行內容的解碼
• 第三階段重新入新的佇列，這將是新的樣本佇列。

注：這些操作需要啟動執行這些排隊操作的執行緒，以便我們的訓練迴圈可以將佇列中的內容入隊出隊操作。

1.1 第一階段

我們稱之為構造檔案佇列，將需要讀取的檔案裝入到一個固定的隊列當中

• tf.train.string_input_producer(string_tensor,shuffle=True)
  • string_tensor：含有檔名+路徑的1階張量
  • num_epochs:過幾遍資料，預設無限過資料
  • return 檔案佇列

1.2、第二階段

這裡需要從隊列當中讀取檔案內容，並且進行解碼操作。關於讀取內容會有一定的規則

1.2.1 讀取檔案內容

TensorFlow預設每次只讀取一個樣本，具體到文字檔案讀取一行、二進位制檔案讀取指定位元組數(最好一個樣本)、圖片檔案預設讀取一張圖片、TFRecords預設讀取一個example

• tf.TextLineReader:
  • 閱讀文字檔案逗號分隔值（CSV）格式,預設按行讀取
  • return：讀取器例項
• tf.WholeFileReader:用於讀取圖片檔案
• tf.TFRecordReader:
  • 讀取TFRecords檔案
• tf.FixedLengthRecordReader:二進位制檔案
  • 要讀取每個記錄是固定數量位元組的二進位制檔案
  • record_bytes:整型，指定每次讀取(一個樣本)的位元組數
  • return：讀取器例項

1、他們有共同的讀取方法：read(file_queue)：從佇列中指定數量內容返回一個Tensors元組（key檔名字，value預設的內容(一個樣本)）

2、由於預設只會讀取一個樣本，所以通常想要進行批處理。使用tf.train.batch或tf.train.shuffle_batch進行多樣本獲取，便於訓練時候指定每批次多個樣本的訓練

1.2.2 內容解碼

對於讀取不通的檔案型別，內容需要解碼操作，解碼成統一的Tensor格式

• tf.decode_csv：解碼文字檔案內容
• tf.decode_raw：解碼二進位制檔案內容
  • 與tf.FixedLengthRecordReader搭配使用，二進位制讀取為uint8格式
• tf.image.decode_jpeg(contents)
  • 將JPEG編碼的影象解碼為uint8張量
  • return:uint8張量，3-D形狀[height, width, channels]
• tf.image.decode_png(contents)
  • 將PNG編碼的影象解碼為uint8張量
  • return:張量型別，3-D形狀[height, width, channels]

解碼階段，預設所有的內容都解碼成tf.uint8格式，如果需要後續的型別處理繼續處理

1.3 第三階段

在解碼之後，我們可以直接獲取預設的一個樣本內容了，但是如果想要獲取多個樣本，這個時候需要結合管道的末尾進行批處理

• tf.train.batch(tensors,batch_size,num_threads = 1,capacity = 32,name=None)
  • 讀取指定大小（個數）的張量
  • tensors：可以是包含張量的列表,批處理的內容放到列表當中
  • batch_size:從佇列中讀取的批處理大小
  • num_threads：進入佇列的執行緒數
  • capacity：整數，佇列中元素的最大數量
  • return:tensors
• tf.train.shuffle_batch

2、執行緒操作

以上的建立這些佇列和排隊操作稱之為tf.train.QueueRunner。每個QueueRunner都負責一個階段，並擁有需要線上程中執行的排隊操作列表。一旦圖形被構建， tf.train.start_queue_runners 函式就會要求圖中的每個QueueRunner啟動它的執行排隊操作的執行緒。（這些操作需要在會話中開啟）

• tf.train.start_queue_runners(sess=None,coord=None)
  • 收集所有圖中的佇列執行緒，並啟動執行緒
  • sess:所在的會話中
  • coord：執行緒協調器
  • return：返回所有執行緒
• tf.train.Coordinator()
  • 執行緒協調員,實現一個簡單的機制來協調一組執行緒的終止
  • request_stop()：請求停止
  • should_stop()：詢問是否結束
  • join(threads=None, stop_grace_period_secs=120)：回收執行緒
  • return:執行緒協調員例項

3、影象基本知識

對於影象檔案，我們怎麼進行轉換成機器學習能夠理解的資料。之前我們講過文字怎麼處理成數字資訊。對於圖片來講，組成圖片的最基本單位是畫素，所以我們獲取的是每張圖片的畫素值。接觸的圖片有兩種，一種是黑白圖片，另一種是彩色圖片。

3.1 圖片三要素

組成一張圖片特徵值是所有的畫素值，有這麼幾個要素。圖片長度、圖片寬度、圖片通道數。什麼是圖片的通道數呢，描述一個畫素點，如果是灰度，那麼只需要一個數值來描述它，就是單通道。如果一個畫素點，有RGB三種顏色來描述它，就是三通道。那所以

• 灰度圖：單通道
• 彩色圖片：三通道

假設一張彩色圖片的長200，寬200，通道數為3，那麼總的畫素數量為200 * 200 * 3

3.2 張量形狀

讀取圖片之後，怎麼用張量形狀來表示呢。一張圖片就是一個3D張量，[height, width, channel]，height就表示高，width表示寬，channel表示通道數。我們會經常遇到3D和4D的表示

• 單個圖片：[height, width, channel]
• 多個圖片：[batch,height, width, channel]，batch表示批數量

3.3 圖片特徵值處理

在進行圖片識別的時候，每個圖片樣本的特徵數量要保持相同。所以需要將所有圖片張量大小統一轉換。另一方面如果圖片的畫素量太大，也可以通過這種方式適當減少畫素的數量，減少訓練的計算開銷

• tf.image.resize_images(images, size)
  • 縮小放大圖片
  • images：4-D形狀[batch, height, width, channels]或3-D形狀的張量[height, width, channels]的圖片資料
  • size：1-D int32張量：new_height, new_width，影象的新尺寸
  • 返回4-D格式或者3-D格式圖片

3.4 資料格式

• 儲存：uint8(節約空間)
• 矩陣計算：float32(提高精度)

4、CIFAR10二進位制資料集介紹

https://www.cs.toronto.edu/~kriz/cifar.html 

• 二進位制版本資料檔案

二進位制版本

二進位制版本包含檔案data_batch_1.bin，data_batch_2.bin，…，data_batch_5.bin以及test_batch.bin

。這些檔案中的每一個格式如下，資料中每個樣本包含了特徵值和目標值：

<1×標籤> <3072×畫素> 
... 
<1×標籤> <3072×畫素>

第一個位元組是第一個影象的標籤，它是一個0-9範圍內的數字。接下來的3072個位元組是影象畫素的值。前1024個位元組是紅色通道值，下1024個綠色，最後1024個藍色。值以行優先順序儲存，因此前32個位元組是影象第一行的紅色通道值。 每個檔案都包含10000個這樣的3073位元組的“行”影象，但沒有任何分隔行的限制。因此每個檔案應該完全是30730000位元組長。

5、TFRecords

5.1 什麼是TFRecords檔案

TFRecords其實是一種二進位制檔案，雖然它不如其他格式好理解，但是它能更好的利用記憶體，更方便複製和移動，並且不需要單獨的標籤檔案。

TFRecords檔案包含了tf.train.Example 協議記憶體塊(protocol buffer)(協議記憶體塊包含了欄位 Features)。可以獲取你的資料， 將資料填入到Example協議記憶體塊(protocol buffer)，將協議記憶體塊序列化為一個字串， 並且通過tf.python_io.TFRecordWriter 寫入到TFRecords檔案。

• 檔案格式 *.tfrecords

5.2 Example結構解析

tf.train.Example 協議記憶體塊(protocol buffer)(協議記憶體塊包含了欄位 Features)，Features包含了一個Feature欄位，Features中包含要寫入的資料、並指明資料型別。這是一個樣本的結構，批資料需要迴圈存入這樣的結構

 example = tf.train.Example(features=tf.train.Features(feature={
                "image": tf.train.Feature(bytes_list=tf.train.BytesList(value=[image])),
                "label": tf.train.Feature(int64_list=tf.train.Int64List(value=[label])),
            }))

• 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])
  • 支援存入的型別如下
  • tf.train.Int64List(value=[Value])
  • tf.train.BytesList(value=[Bytes])
  • tf.train.FloatList(value=[value])

這種結構是不是很好的解決了資料和標籤(訓練的類別標籤)或者其他屬性資料儲存在同一個檔案中

完整Demo

# -*- coding=utf-8 -*-
import os
# os.environ["TF_CPP_MIN_LOG_LEVEL"]='1' # 這是預設的顯示等級，顯示所有資訊  
os.environ["TF_CPP_MIN_LOG_LEVEL"]='2' # 只顯示 warning 和 Error   
# os.environ["TF_CPP_MIN_LOG_LEVEL"]='3' # 只顯示 Error

import tensorflow as tf


def picread(file_list):
    """
    讀取狗圖片資料到張量
    :param file_list:路徑+檔名到列表
    :return:
    """
    # 1. 構建檔案佇列
    file_queue = tf.train.string_input_producer(file_list)

    # 2. 利用圖片讀取器去讀取檔案佇列內容
    reader = tf.WholeFileReader()

    # 3. 預設一次一張圖片,沒有形狀
    key, value = reader.read(file_queue)
    print(value)

    # 4. 對圖片資料進行解碼
    # string --> unit8
    # () ---> (?, ?, ?)
    image = tf.image.decode_jpeg(value)
    print(image)

    # 5. 圖片的形狀固定、大小處理
    # 把影象大小固定統一大小(演算法訓練要求樣本的特徵值數量一樣)
    # 固定【200, 200]
    image_resize = tf.image.resize_images(image, size=[200, 200])
    print(image_resize)

    # 6. 設定圖片形狀
    image_resize.set_shape([200, 200, 3])

    # 7. 進行批處理
    # 3D --> 4D
    # tf.train.batch(tensors,batch_size,num_threads = 1,capacity = 32,name=None)
    #   - 讀取指定大小（個數）的張量
    #   - tensors：可以是包含張量的列表,批處理的內容放到列表當中
    #   - batch_size:從佇列中讀取的批處理大小
    #   - num_threads：進入佇列的執行緒數
    #   - capacity：整數，佇列中元素的最大數量
    #   return:tensors
    image_batch = tf.train.batch([image_resize], batch_size=10, num_threads=1, capacity=8)
    print(image_batch)

    return image_batch


class Cifarread(object):
    """
    讀取cifar10類別的二進位制檔案
    """

    def __init__(self):
        # 每個影象樣本的屬性
        self.height = 32
        self.width = 32
        self.channel = 3

        # bytes
        # 標籤位元組
        self.label_bytes = 1
        # 特徵值位元組
        self.image_bytes = self.height * self.width * self.channel
        # 總的位元組數
        self.all_bytes = self.label_bytes + self.image_bytes

    def bytes_read(self, file_list):
        """
        讀取二進位制解碼張量
        :param file_list: 路徑+檔名到列表
        :return:
        """
        # 1. 構造檔案佇列
        file_queue = tf.train.string_input_producer(file_list)

        # 2. 使用tf.FixedLengthRecordReader(bytes)讀取
        # 預設必須指定讀取一個樣本
        reader = tf.FixedLengthRecordReader(self.all_bytes)

        # 讀取佇列
        _, value = reader.read(file_queue)

        # 3. 解碼操作
        # (?, )  (3.73, )  =  label(1, ) + feature(3072, )
        label_image = tf.decode_raw(value, tf.uint8)
        # 為了訓練方便，一般會把特徵值和目標值分開處理
        print(label_image)

        # 使用tf.slice進行切片,tf.cast更改所要的型別
        label = tf.cast(tf.slice(label_image, [0], [self.label_bytes]), tf.int32)  # tf.cast：將原來的unit8 改為 int32型別
        image = tf.slice(label_image, [self.label_bytes], [self.image_bytes])

        print(label, image)

        # 處理型別和圖片資料的形狀
        # 圖片形狀[32, 32, 3]
        # reshape (3072, )  --> [channel, height, width]
        # transpose [channel, height, width] --> [height, width, channel]
        depth_major = tf.reshape(image, [self.channel, self.height, self.width])
        print(depth_major)
        image_reshape = tf.transpose(depth_major, [1, 2, 0])
        print(image_reshape)

        # 4. 批處理
        # 3D --> 4D
        # tf.train.batch(tensors,batch_size,num_threads = 1,capacity = 32,name=None)
        #   - 讀取指定大小（個數）的張量
        #   - tensors：可以是包含張量的列表,批處理的內容放到列表當中
        #   - batch_size:從佇列中讀取的批處理大小
        #   - num_threads：進入佇列的執行緒數
        #   - capacity：整數，佇列中元素的最大數量
        #   return:tensors
        image_batch, label_batch = tf.train.batch([image_reshape, label], batch_size=10, num_threads=1, capacity=10)

        return image_batch, label_batch

    def write_to_tfrecords(self, image_batch, label_batch):
        """
        將資料寫入TFRecords檔案
        :param image_batch: 特徵值
        :param label_batch: 目標值
        :return:
        """
        # 構造TFRecords儲存器
        writer = tf.python_io.TFRecordWriter("./temp/cifar.tfrecords")

        #  迴圈將每個樣本構造成一個example，然後序列化寫入
        for i in range(10):
            # 取出相應對第i個樣本的特徵值和目標值
            # 寫入的是具體的張量的值，不是OP的名字
            # [10, 32, 32, 3]  --> [32, 32, 3]
            image = image_batch[i].eval().tostring()  # 將其轉化為Bytes型別
            #[10, 1]
            label = int(label_batch[i].eval()[0])

            # 每個樣本對example
            example = tf.train.Example(features=tf.train.Features(feature={
                "image": tf.train.Feature(bytes_list=tf.train.BytesList(value=[image])),
                "label": tf.train.Feature(int64_list=tf.train.Int64List(value=[label]))
            }))

            # 寫入第i個樣本的example
            writer.write(example.SerializeToString())
        writer.close()
        return None

    def read_tfrecords(self):
        """
        讀取tfrecords的資料
        :return: None
        """
        # 1、構造檔案佇列
        file_queue = tf.train.string_input_producer(["./temp/cifar.tfrecords"])

        # 2、構造tfrecords讀取器，讀取佇列
        reader = tf.TFRecordReader()

        # 預設也是隻讀取一個樣本
        key, values = reader.read(file_queue)

        # tfrecords
        # 多瞭解析example的一個步驟
        feature = tf.parse_single_example(values, features={
            "image": tf.FixedLenFeature([], tf.string),
            "label": tf.FixedLenFeature([], tf.int64)
        })

        # 取出feature裡面的特徵值和目標值
        # 通過鍵值對獲取
        image = feature["image"]

        label = feature["label"]

        # 3、解碼操作
        # 對於image是一個bytes型別，所以需要decode_raw去解碼成uint8張量
        # 對於Label:本身是一個int型別，不需要去解碼
        image = tf.decode_raw(image, tf.uint8)

        print(image, label)

        # # 從原來的[32,32,3]的bytes形式直接變成[32,32,3]
        # 不存在一開始我們的讀取RGB的問題
        # 處理image的形狀和型別
        image_reshape = tf.reshape(image, [self.height, self.width, self.channel])

        # 處理label的形狀和型別
        label_cast = tf.cast(label, tf.int32)

        print(image_reshape, label_cast)

        # 4、批處理操作
        image_batch, label_batch = tf.train.batch([image_reshape, label_cast], batch_size=10, num_threads=1, capacity=10)

        print(image_batch, label_batch)
        return image_batch, label_batch


if __name__ == '__main__':
    # 例1：影象資料讀取
    filename = os.listdir("../data/dog/")
    file_list = [os.path.join("../data/dog/", file) for file in filename]
    # 例 2：二進位制資料讀取
    filename_01 = os.listdir("../data/cifar10/cifar-10-batches-bin/")
    file_list_01 = [os.path.join("../data/cifar10/cifar-10-batches-bin/", file) for file in filename_01 if file[-3:] == "bin"]

    # print(file_list)

    # 0. 傳入資料
    # 例1：
    image_batch = picread(file_list)
    # 例2：
    cr = Cifarread()
    image_batch_01, label_batch_01 = cr.bytes_read(file_list_01)
    # 例3：讀取TFRecords的結果
    image_batch_02, label_batch_02 = cr.read_tfrecords()

    with tf.Session() as sess:
        # 8.建立執行緒回收的協調器
        # coord = tf.train.Coordinator()
        # 執行緒協調員, 實現一個簡單的機制來協調一組執行緒的終止
        # coord.request_stop()：請求停止
        # coord.should_stop()：詢問是否結束
        # coord.join(threads=None, stop_grace_period_secs=120)：回收執行緒
        # return:執行緒協調員例項
        coord = tf.train.Coordinator()

        # 9. 需要手動開啟子執行緒去進行批量處理讀取到佇列操作
        # tf.train.start_queue_runners(sess=None, coord=None)
        # 收集所有圖中的佇列執行緒，並啟動執行緒
        #   - sess: 所在的會話中
        #   - coord：執行緒協調器
        # return：返回所有執行緒
        threads = tf.train.start_queue_runners(sess=sess, coord=coord)

        # ⚠️列印內容、執行：需要開啟子執行緒去執行，子執行緒就把資料讀取到佇列，主執行緒取出資料去訓練
        print("image:", sess.run(image_batch))
        print("Bytes:", sess.run([image_batch_01, label_batch_01]))
        print(".tfrecords:", sess.run([image_batch_02, label_batch_02]))

        # 寫入檔案
        # cr.write_to_tfrecords(image_batch_01, label_batch_01)

        # 10.回收執行緒
        coord.request_stop()
        coord.join(threads)

    with tf.Session() as sess:
        # 8.建立執行緒回收的協調器
        # coord = tf.train.Coordinator()
        # 執行緒協調員, 實現一個簡單的機制來協調一組執行緒的終止
        # coord.request_stop()：請求停止
        # coord.should_stop()：詢問是否結束
        # coord.join(threads=None, stop_grace_period_secs=120)：回收執行緒
        # return:執行緒協調員例項
        coord = tf.train.Coordinator()

        # 9. 需要手動開啟子執行緒去進行批量處理讀取到佇列操作
        # tf.train.start_queue_runners(sess=None, coord=None)
        # 收集所有圖中的佇列執行緒，並啟動執行緒
        #   - sess: 所在的會話中
        #   - coord：執行緒協調器
        # return：返回所有執行緒
        threads = tf.train.start_queue_runners(sess=sess, coord=coord)

        # ⚠️列印內容、執行：需要開啟子執行緒去執行，子執行緒就把資料讀取到佇列，主執行緒取出資料去訓練
        print("image:", sess.run(image_batch))
        print("Bytes:", sess.run([image_batch_01, label_batch_01]))
        # print(".tfrecords:", sess.run([image_batch_02, label_batch_02]))

        # 寫入檔案
        cr.write_to_tfrecords(image_batch_01, label_batch_01)
        print("news_writer")

        # 10.回收執行緒
        coord.request_stop()
        coord.join(threads)