TensorFlow基礎(chǔ)3：數(shù)據(jù)讀取的三種方式

碼農(nóng)書館 2020-09-30

展開全文

‘在講述在TensorFlow上的數(shù)據(jù)讀取方式之前，有必要了解一下TensorFlow的系統(tǒng)架構(gòu),，如下圖所示：

TensorFlow系統(tǒng)架構(gòu)圖

TensorFlow的系統(tǒng)架構(gòu)分為兩個部分：

前端系統(tǒng)：提供編程模型,，負責(zé)構(gòu)造計算圖；
后端系統(tǒng)：提供運行時環(huán)境,，負責(zé)執(zhí)行計算圖,。

在處理數(shù)據(jù)的過程當(dāng)中,，由于現(xiàn)在的硬件性能的極大提升，數(shù)值計算過程可以通過加強硬件的方式來改善,，因此數(shù)據(jù)讀?。碔O)往往會成為系統(tǒng)運行性能的瓶頸。在TensorFlow框架中提供了三種數(shù)據(jù)讀取方式：

Preloaded data: 預(yù)加載數(shù)據(jù)
Feeding: placeholder, feed_dict 由占位符代替數(shù)據(jù),，運行時填入數(shù)據(jù)
Reading from file: 從文件中直接讀取

以上三種讀取方式各有自己的特點,，在了解這些特點或區(qū)別之前，需要知道TensorFlow是如何進行工作的,。

TF的核心是用C++寫的,，這樣的好處是運行快，缺點是調(diào)用不靈活,。而Python恰好相反,，所以結(jié)合兩種語言的優(yōu)勢。涉及計算的核心算子和運行框架是用C++寫的,，并提供API給Python,。Python調(diào)用這些API，設(shè)計訓(xùn)練模型(Graph),，再將設(shè)計好的Graph給后端去執(zhí)行,。簡而言之，Python的角色是Design,，C++是Run,。

1. Preload data: constant 預(yù)加載數(shù)據(jù)

特點：數(shù)據(jù)直接嵌入graph，由graph傳入session中運行

import tensorflow as tf

#設(shè)計graph
x = tf.constant([1,2,3], name='x')
y = tf.constant([2,3,4], name='y')
z = tf.add(x,y, name='z')

#打開一個session,，計算z
with tf.Session() as sess:
    print(sess.run(z))

運行結(jié)果如下：
這里寫圖片描述
在設(shè)計Graph的時候,，x和y就被定義成了兩個有值的列表，在計算z的時候直接取x1和x2的值,。

2.Feeding: placeholder, feed_dict

特點：由占位符代替數(shù)據(jù),，運行時填入數(shù)據(jù)

import tensorflow as tf

#設(shè)計graph，用占位符代替
x = tf.placeholder(tf.int16)
y = tf.placeholder(tf.int16)
z = tf.add(x,y, name='z')

#打開一個session
with tf.Session() as sess:
    #創(chuàng)建數(shù)據(jù)
    xs = [1,2,3]
    ys = [2,3,4]
    #運行session,用feed_dict來將創(chuàng)建的數(shù)據(jù)傳遞進占位符
    print(sess.run(z, feed_dict={x: xs, y: ys}))

運行結(jié)果如下：
這里寫圖片描述
在這里x, y只是占位符,，沒有具體的值,，那么運行的時候去哪取值呢？這時候就要用到sess.run()中的feed_dict參數(shù),，將Python產(chǎn)生的數(shù)據(jù)喂給后端,，并計算z。

3.Reading From File：直接從文件中讀取

前兩種方法很方便,，但是遇到大型數(shù)據(jù)的時候就會很吃力,，即使是Feeding，中間環(huán)節(jié)的增加也是不小的開銷，比如數(shù)據(jù)類型轉(zhuǎn)換等等,。最優(yōu)的方案就是在Graph定義好文件讀取的方法,，讓TF自己去從文件中讀取數(shù)據(jù)，并解碼成可使用的樣本集,。

這種直接從文件中讀取數(shù)據(jù)的方式需要設(shè)計成Queue的方式才能較好的解決IO瓶頸的問題,。
Queue機制有如下三個特點：

producer-consumer pattern(生產(chǎn)消費模式)
獨立于主線程執(zhí)行
異步IO: reader.read(queue) tf.train.batch()

這里寫圖片描述

在上圖中，首先由一個單線程把文件名堆入隊列,，兩個Reader同時從隊列中取文件名并讀取數(shù)據(jù),，Decoder將讀出的數(shù)據(jù)解碼后堆入樣本隊列，最后單個或批量取出樣本（圖中沒有展示樣本出列）,。我們這里通過三段代碼逐步實現(xiàn)上圖的數(shù)據(jù)流,，這里我們不使用隨機，讓結(jié)果更清晰,。

文件準備

$ echo -e "Alpha1,A1\nAlpha2,A2\nAlpha3,A3" > A.csv  
$ echo -e "Bee1,B1\nBee2,B2\nBee3,B3" > B.csv  
$ echo -e "Sea1,C1\nSea2,C2\nSea3,C3" > C.csv  
$ cat A.csv  
Alpha1,A1  
Alpha2,A2  
Alpha3,A3

單個Reader,，單個樣本

import tensorflow as tf  
# 生成一個先入先出隊列和一個QueueRunner  
filenames = ['A.csv', 'B.csv', 'C.csv']  
filename_queue = tf.train.string_input_producer(filenames, shuffle=False)  
# 定義Reader  
reader = tf.TextLineReader()  
key, value = reader.read(filename_queue)  
# 定義Decoder  
example, label = tf.decode_csv(value, record_defaults=[['null'], ['null']])  
# 運行Graph  
with tf.Session() as sess:  
    coord = tf.train.Coordinator()  #創(chuàng)建一個協(xié)調(diào)器，管理線程  
    threads = tf.train.start_queue_runners(coord=coord)  #啟動QueueRunner, 此時文件名隊列已經(jīng)進隊,。  
    for i in range(10):  
        print example.eval()   #取樣本的時候,，一個Reader先從文件名隊列中取出文件名，讀出數(shù)據(jù),，Decoder解析后進入樣本隊列,。  
    coord.request_stop()  
    coord.join(threads)  
# outpt  
Alpha1  
Alpha2  
Alpha3  
Bee1  
Bee2  
Bee3  
Sea1  
Sea2  
Sea3  
Alpha1

單個Reader，多個樣本

import tensorflow as tf  
filenames = ['A.csv', 'B.csv', 'C.csv'] 
## filenames = tf.train.match_filenames_once('.\data\*.csv') 
filename_queue = tf.train.string_input_producer(filenames, shuffle=False)  
reader = tf.TextLineReader()  
key, value = reader.read(filename_queue)  
example, label = tf.decode_csv(value, record_defaults=[['null'], ['null']])  
# 使用tf.train.batch()會多加了一個樣本隊列和一個QueueRunner,。Decoder解后數(shù)據(jù)會進入這個隊列,，再批量出隊。  
# 雖然這里只有一個Reader,，但可以設(shè)置多線程,，相應(yīng)增加線程數(shù)會提高讀取速度，但并不是線程越多越好,。  
example_batch, label_batch = tf.train.batch(  
      [example, label], batch_size=5)  
with tf.Session() as sess:  
    coord = tf.train.Coordinator()  
    threads = tf.train.start_queue_runners(coord=coord)  
    for i in range(10):  
        print example_batch.eval()  
    coord.request_stop()  
    coord.join(threads)  
# output  
# ['Alpha1' 'Alpha2' 'Alpha3' 'Bee1' 'Bee2']  
# ['Bee3' 'Sea1' 'Sea2' 'Sea3' 'Alpha1']  
# ['Alpha2' 'Alpha3' 'Bee1' 'Bee2' 'Bee3']  
# ['Sea1' 'Sea2' 'Sea3' 'Alpha1' 'Alpha2']  
# ['Alpha3' 'Bee1' 'Bee2' 'Bee3' 'Sea1']  
# ['Sea2' 'Sea3' 'Alpha1' 'Alpha2' 'Alpha3']  
# ['Bee1' 'Bee2' 'Bee3' 'Sea1' 'Sea2']  
# ['Sea3' 'Alpha1' 'Alpha2' 'Alpha3' 'Bee1']  
# ['Bee2' 'Bee3' 'Sea1' 'Sea2' 'Sea3']  
# ['Alpha1' 'Alpha2' 'Alpha3' 'Bee1' 'Bee2']

多Reader,，多個樣本

import tensorflow as tf  
filenames = ['A.csv', 'B.csv', 'C.csv']  
filename_queue = tf.train.string_input_producer(filenames, shuffle=False)  
reader = tf.TextLineReader()  
key, value = reader.read(filename_queue)  
record_defaults = [['null'], ['null']]  
example_list = [tf.decode_csv(value, record_defaults=record_defaults)  
                  for _ in range(2)]  # Reader設(shè)置為2  
# 使用tf.train.batch_join()，可以使用多個reader,，并行讀取數(shù)據(jù),。每個Reader使用一個線程,。  
example_batch, label_batch = tf.train.batch_join(  
      example_list, batch_size=5)  
with tf.Session() as sess:  
    coord = tf.train.Coordinator()  
    threads = tf.train.start_queue_runners(coord=coord)  
    for i in range(10):  
        print example_batch.eval()  
    coord.request_stop()  
    coord.join(threads)  

# output  
# ['Alpha1' 'Alpha2' 'Alpha3' 'Bee1' 'Bee2']  
# ['Bee3' 'Sea1' 'Sea2' 'Sea3' 'Alpha1']  
# ['Alpha2' 'Alpha3' 'Bee1' 'Bee2' 'Bee3']  
# ['Sea1' 'Sea2' 'Sea3' 'Alpha1' 'Alpha2']  
# ['Alpha3' 'Bee1' 'Bee2' 'Bee3' 'Sea1']  
# ['Sea2' 'Sea3' 'Alpha1' 'Alpha2' 'Alpha3']  
# ['Bee1' 'Bee2' 'Bee3' 'Sea1' 'Sea2']  
# ['Sea3' 'Alpha1' 'Alpha2' 'Alpha3' 'Bee1']  
# ['Bee2' 'Bee3' 'Sea1' 'Sea2' 'Sea3']  
# ['Alpha1' 'Alpha2' 'Alpha3' 'Bee1' 'Bee2']

tf.train.batch與tf.train.shuffle_batch'數(shù)是單個Reader讀取,，但是可以多線程。tf.train.batch_join'和tf.train.shuffle_batch_join可設(shè)置多Reader讀取,，每個Reader使用一個線程,。至于兩種方法的效率，單Reader時，2個線程就達到了速度的極限,。多Reader時,，2個Reader就達到了極限。所以并不是線程越多越快,，甚至更多的線程反而會使效率下降,。

迭代控制

filenames = ['A.csv', 'B.csv', 'C.csv']  
filename_queue = tf.train.string_input_producer(filenames, shuffle=False, num_epochs=3)  # num_epoch: 設(shè)置迭代數(shù)  
reader = tf.TextLineReader()  
key, value = reader.read(filename_queue)  
record_defaults = [['null'], ['null']]  
example_list = [tf.decode_csv(value, record_defaults=record_defaults)  
                  for _ in range(2)]  
example_batch, label_batch = tf.train.batch_join(  
      example_list, batch_size=5)  
init_local_op = tf.initialize_local_variables()  
with tf.Session() as sess:  
    sess.run(init_local_op)   # 初始化本地變量   
    coord = tf.train.Coordinator()  
    threads = tf.train.start_queue_runners(coord=coord)  
    try:  
        while not coord.should_stop():  
            print example_batch.eval()  
    except tf.errors.OutOfRangeError:  
        print('Epochs Complete!')  
    finally:  
        coord.request_stop()  
    coord.join(threads)  
    coord.request_stop()  
    coord.join(threads)  
# output  
# ['Alpha1' 'Alpha2' 'Alpha3' 'Bee1' 'Bee2']  
# ['Bee3' 'Sea1' 'Sea2' 'Sea3' 'Alpha1']  
# ['Alpha2' 'Alpha3' 'Bee1' 'Bee2' 'Bee3']  
# ['Sea1' 'Sea2' 'Sea3' 'Alpha1' 'Alpha2']  
# ['Alpha3' 'Bee1' 'Bee2' 'Bee3' 'Sea1']  
# Epochs Complete!