這一節我們提及了三個內容：變數共享、執行緒和佇列和資料讀取，這些都是TensorFlow官方指導中的內容。會在程式中經常遇到所以放在一起進行敘述。前面都是再利用已有的資料進行tensorflow的學習，這一節我們要學習怎麼從檔案中讀取我們需要的各類資料。

1、變數共享

　　前面已經說過如何進行變數的生成和初始化內容，也用到了名稱空間的概念，這裡說一下什麼是變數共享。當我們有一個非常龐大的模型的時候免不了需要進行大量的變數共享，而且有時候還希望能夠在一個地方初始化所有的變數，這就需要tf.variable_scope() 和 tf.get_variable()。
　　當只有兩層的卷積的時候，前面的程式都是定義了兩個卷積變數W1和W2(忽略b)，而且簡單的命名了。如下面:

def my_image_filter(input_images):
    conv1_weights = tf.Variable(tf.random_normal([5, 5, 32, 32]),
        name="conv1_weights")
    conv1_biases = tf.Variable(tf.zeros([32]), name="conv1_biases")
    conv1 = tf.nn.conv2d(input_images, conv1_weights,
        strides=[1, 1, 1, 1], padding='SAME'
 
)
    relu1 = tf.nn.relu(conv1 + conv1_biases)

    conv2_weights = tf.Variable(tf.random_normal([5, 5, 32, 32]),
        name="conv2_weights")
    conv2_biases = tf.Variable(tf.zeros([32]), name="conv2_biases")
    conv2 = tf.nn.conv2d(relu1, conv2_weights,
        strides=[1, 1, 1, 1], padding='SAME'
 
)
    return tf.nn.relu(conv2 + conv2_biases)

　　但是假設你在reuse這個模型的時候如果有兩個影象都呼叫my_image_filter就會有4*2個變數產生。用全域性變數進行引數表示當然可以，但是卻破壞了程式碼封裝性，於是用tensorflow的變數作用域（Variable Scope）概念可以解決這個問題。這不得不用到兩個函式：

• tf.get_variable(name, shape, initializer): Creates or returns a variable with a given name.建立並返回一個給定名稱的變數

• tf.variable_scope( scope_name): Manages namespaces for names passed to tf.get_variable(). 管理變數名稱空間忽略get_variable定義的變數，而name_scope是所有的都會放到其名稱空間中見這裡

  　　如何用這兩個方式實現變數共享呢？方法如下假設被呼叫的函式為：

　　def my_image_filter(input_images):
    with tf.variable_scope("conv1"):
        # Variables created here will be named "conv1/weights", "conv1/biases".
        relu1 = conv_relu(input_images, [5, 5, 32, 32], [32])
    with tf.variable_scope("conv2"):
        # Variables created here will be named "conv2/weights", "conv2/biases".
        return conv_relu(relu1, [5, 5, 32, 32], [32])

只要用到reuse_variables()進行註釋即可，方法如下：

with tf.variable_scope("image_filters") as scope:
    result1 = my_image_filter(image1)
    scope.reuse_variables()
    result2 = my_image_filter(image2)

至於為什麼會這樣，建議參考官方文件，因為後面還會用到所以這裡只是大概瞭解一下。

2、執行緒(Threading)和佇列(queues)

　　這個是一個基本概念，也是後續資料讀取的基礎。這裡會對相關概念進行一下詳細的分析，包括佇列、協調器（Coordinator）、QueueRunner。

（1）執行緒和佇列使用的總覽

　　在TensorFlow中使這麼描寫佇列的:Queues are a powerful mechanism for asynchronous(非同步) computation using TensorFlow。佇列其實也是一個節點，是個有點類似於variable的狀態節點，通過入隊（enqueue）和出隊（dequeue）來維持一種狀態。其中非同步是一個很重要的內容，以資料讀取為例：在輸入的pipeline上，多執行緒可以有效地增加資料讀取的效率，當我們用佇列來準備用於訓練的資料時可以：
　　* 多執行緒的準備訓練資料，然後把它們push到佇列中
　　* 訓練執行緒生成一個一個訓練operation來從資料佇列中去除mini-batch數量的資料。
　　其實Session就是多執行緒的，所以TensorFlow可以並行執行，但是多執行緒有三個條件需要滿足:
　　- 所有執行緒都需要能夠同時停止
　　- 必須能夠捕捉和報告異常
　　- 佇列必須能在停止的時候適當關閉
　　TensorFlow給了我們兩個類用於幫助我們協調執行緒：tf.Coordinator和tf.train.QueueRunner。這兩個類實際上是一起使用的，其中tf.Coordinator用於多執行緒停止和異常捕捉，tf.train.QueueRunner用於操作Tensor入隊。

（2）主要的佇列類

上面對執行緒和佇列的作用進行了一些簡單的介紹，我們常用的佇列在TensorFlow中已經定義好了：tf.FIFOQueue和tf.RandomShuffleQueue，從名字就可以瞭解到兩個佇列的工作方式一個按順序一個隨機。這兩個佇列支援enqueue、enqueue_many和dequeue操作，這些操作內容也能直接理解出來。最長見的操作就是一次性讀入資料，然後一個一個的出隊，這裡注意是沒有dequeue_many操作的，如果想一次性得到多個輸出需要使用tf.train.batch或是tf.train.shuffle_batch。
　　這裡還有一個在sequence-to-sequence上比較常用的tf.PaddingFIFOQueue也就是帶變數填充的佇列，而且它支援dequeue_many。
　　我們可以自己定義一個佇列，其常用的引數有min_after_dequeue（在dequeue後佇列中剩餘的最小數目的元素）、bounded capacity（佇列中的最大元素數）、shape of the elements in the queue（如果shape為None則元素可以是任何shape）。但是在實際中我們通常不會直接使用佇列本身，而是使用string_input_producer（後面讀取資料時會用到）。

（2）Coordinator

　　前面說過協調器幫助多執行緒共同停止。其關鍵的方法有:

• tf.train.Coordinator.should_stop：如果執行緒可以停止返回返回True
• tf.train.Coordinator.request_stop ：　請求多執行緒停止
• tf.train.Coordinator.join：直到指定執行緒停止前進行等待

我們可以初始化一個Coordinator物件，然後製造一些執行緒進行測試。教程給的程式碼如下,一個執行緒申請申請停止，其他執行緒的should_stop就會變為True：

# Thread body: loop until the coordinator indicates a stop was requested.
# If some condition becomes true, ask the coordinator to stop.
def MyLoop(coord):
  while not coord.should_stop():
    ...do something...
    if ...some condition...:
      coord.request_stop()

# Main thread: create a coordinator.
coord = tf.train.Coordinator()

# Create 10 threads that run 'MyLoop()'這個是Python package創造的執行緒
threads = [threading.Thread(target=MyLoop, args=(coord,)) for i in xrange(10)]

# Start the threads and wait for all of them to stop.
for t in threads:
  t.start()#啟動執行緒
coord.join(threads)

　　顯然coordinator可以管理許多執行緒做不同的事情，而不用像上面的例子那樣都做同一件事。而且關於異常的捕獲和丟擲可以看tf.train.Coordinator的文件。

（4）QueueRunner

　　QueueRunne類會生成一些執行緒反覆的執行enqueue操作。然後這些執行緒可以利用Coordinator來停止這些執行緒。而且如果有異常報道至Coordinator時一個QueueRunner會自動執行一個closer thread來關閉執行緒。
　　下面可以自己使用queue runner來執行上面的結構：
　　1. 首先建立一個使用TensorFlow佇列（如tf.RandomShuffleQueue）的圖，接著增加一個Op用於處理資料並將其入隊。
　　2. 建立一個queuerunner並行的執行4個執行緒進行入隊，然後建立一個Coordinator讓QueueRunner在這個Coordinator上開始執行緒，並在執行期間不斷使用Coordinator監控停止情況。
　　這裡沒有用官方文件，因為裡面有些東西有一些錯誤。用的是這裡的程式。

N_SAMPLES = 1000#樣本數
NUM_THREADS = 4#執行緒數
# Generating some simple data生成資料和標籤
# create 1000 random samples, each is a 1D array from the normal distribution (10, 1)
data = 10 * np.random.randn(N_SAMPLES, 4)+1
# create 1000 random labels of 0 and 1
target = np.random.randint(0, 2, size=N_SAMPLES)
#建立一個佇列
queue = tf.FIFOQueue(capacity = 50, dtypes=[tf.float32, tf.int32 ], shapes =[[4], []])
#增加Op用於資料入隊出隊
enqueue_op = queue.enqueue_many([data, target])
dequeue_op = queue.dequeue()
# create NUM_THREADS to do enqueue建立一個QueueRunner用於管理4個執行緒
qr = tf.train.QueueRunner(queue, [enqueue_op]*NUM_THREADS)
with tf.Session() as sess:
    # Create a coordinator, launch the queue runner threads.生成一個Coordinator啟動執行緒
    coord = tf.train.Coordinator()
    enqueue_threads = qr.create_threads(sess, coord=coord, start=True)
    for step in xrange (100): #do to 100 iterations，執行100次迴圈進行出隊操作，得到需要的資料
        if coord.should_stop():
            break
            data_batch,label_batch=sess.run(dequeue_op)#這裡是一條一條的資料
    coord.request_stop()
    coord.join (enqueue_threads)

3、資料讀取

　　我們前面讀取資料都是封裝好的內容了，下面就開啟封裝看看tensorflow是如何讀取資料的。上面的佇列和執行緒這一部分內容是資料讀取的基礎內容，可以兩部分對照著看。資料讀取有三個來源：Python程式碼提供的（feeding）、從檔案中讀出來的、預載入資料。我們的內容主要集中在檔案讀取上。

（1）feeding

　　這個我們遇到了幾次，前面一直用的這種方法，先用python讀資料再給tensorflow，需要配合Placeholder。python讀資料可以用pickle結合os讀取資料（這個看看Python就可以了）。但是這種方式讀取資料卻會出現一個問題就是資料是先讀到我們的client手中然後再傳入需要資料的地方（workers），當client和worker不在同一個機器上時就會非常慢，這種需要兩次傳導肯定會比較慢，可能單機的時候沒那麼明顯而已。

（1）從檔案讀入

　　從檔案讀入可以避免上面的問題，可以直接從資料儲存的地方讀到worker中。
　　這裡有三個blog1(這個對資料讀取的概念講的很好)、blog2、blog3講的比較詳細，我根據官方文件進行了一些補充。
細，我根據官方文件進行了一些補充。

1. 首先建立一個待讀入的檔名列表。
   有下面幾種定義形式：[‘file0’, ‘file1’]、[(“file%d” % i) for i in range(2)]，或者直接用python的os生成。
2. 建立檔名佇列
   用tf.train.string_input_producer建立一個檔名佇列，其有兩個引數shffle（是否打亂檔名）和num_epoch(epoch需要的數量)。
3. 定義一個reader
   需要定義一個reader以便在檔名佇列中讀取資料。這時候由於有不同的資料格式存在，可以選擇不同的reader，其中TextLineReader用的相對較多。而可以把Reader想象成一個每次返回不同值的Op（有點類似於Python的生成器）,當我們使用Reader.read()的時候返回鍵值對，鍵用於鑑定檔名，值為對應的資料值。
   
   • csv或txt格式：使用tf.TextLineReader，配合tf.decode_csv進行讀取。每次讀取一行，其中tf.decode_csv為一個op，把讀出來的資料放入一個tensor列表中（其record_defaults引數用於填充缺失值）
   • bin格式：對於二進位制格式可以用tf.FixedLengthRecordReader讀取，在所有的檔案都是固定長度的時候讀取整個檔案，可以配合tf.decode_raw操作（將一個string轉為uint8的tensor），比如圖片資料就可以這麼讀取，見這裡
   • 標準tensorflow格式：不論什麼資料都轉為tensorflow的支援格式，推薦的格式是TFRecords file，然後再用tf.TFRecordReader進行讀取，用 tf.parse_single_example進行解碼，後面會提到。
   • 一次讀取全部資料：tf.WholeFileReader，其read(queue, name=None)方法返回的(key, value)。key為檔名，value為檔案對應的內容。有一個例子可以參加這裡和這裡。
   • 這裡還有一個tf . ReaderBase幫助我們自定義reader。

這裡以一個csv讀取為例，我們命名為例1,完整程式見這裡（源程式有一些函式已經過時如pack已經變為stack）：

# 例1
filename_queue=tf.train.string_input_producer(["heart.csv"])
reader=tf.TextLineReader(skip_header_lines=1)
# it means you choose to skip the first line for every file in the queue
# 跳過第一行，比如第一行是一些特徵名。
key,value=reader.read(filename_queue)

讀出來的資料形式可能為：

key = data/heart.csv:2
value = 144, 0.01, 4.41, 28.61, Absent, 55, 28.87, 2.06, 63,1

　　train.string_input_producer所建立的佇列是一個封裝好的FIFOQueue用於讀取檔名，這部分的流程最好參考一下上面提到的blog1。然後產生一個Reader這相當於資料佇列，所以為了執行這個佇列需要tf.Coordinator 和 tf.QueueRunner。程式如下：

# 例1 續
filename_queue =tf.train.string_input_producer(filenames)
reader = tf.TextLineReader(skip_header_lines=1) # skip the first line in the file
key, value = reader.read(filename_queue)
with tf.Session() as sess:
    coord = tf.train.Coordinator ()
    threads = tf.train.start_queue_runners(coord=coord,sess=sess)#這個相當於前面的定義執行緒並開啟
    print sess.run(key) # data/heart.csv:2
    print sess.run(value) # 144, 0.01, 4.41, 28.61, Absent, 55, 28.87, 2.06, 63,1
    coord.request_stop ()
    coord.join(threads)

　　上面程式得到的value其實是一個string型別的，為了能得到可用的資料，比如上面的value包含9個特徵和1個lable，我們需要使用decoder,decoder通常有兩個引數，第一個是資料，第二個是預設資料。其中預設資料有兩個作用，一個是指定value中的資料型別，還有一個就是在value中某一列為空時進行填充：

content=tf.decode_csv(value,record_defaults=record_defaults)

現在繼續上面的例子：

# 例1 續
record_defaults = [[1.0] for _ in range(N_FEATURES)] # define all features to be floats
record_defaults[4] = [''] # make the fifth feature string，第四個特徵為string型別
record_defaults.append([1])# 讓標籤變為整數
content=tf.decode_csv(value,record_defaults=record_defaults)

（3）預處理

　　可以對上面產生的資料進行一下預處理。比如資料增強，列表切分等等。這裡繼續對上面的例子進行預處理，比如對某些資料進行處理並切割資料。

# 例1續
# convert the 5th column (present/absent) to the binary value 0 and 1
# 轉換第5列中的string為一個0.0或1.0的常數值
condition=tf.equal(content[4],tf.constant('Present')) #判斷是否出席返回false或true
content[4] = tf.select(condition,tf.constant(1.0),tf.constant(0.0))#這個的意思是對應位置為true就變為1.0
# pack all 9 features into a tensor
features=tf.stack(content[:N_FEATURES]) #這個函式用於打包tf.stack([x, y, z]) = np.asarray([x, y, z])
# assign the last column to label
label=content[-1]

（4）Batching

在結束pipeline之前需要另一個佇列來把資料整合起來，可以利用tf.train.batch或tf.train.shuffle_batch。程式如下：

# 例1
# minimum number elements in the queue after a dequeue, used to ensure
# that the samples are sufficiently mixed
# I think 10 times the BATCH_SIZE is sufficient
# 10倍的batch_size作為出隊後，隊中最少的資料量，越大shuffle約充分但是速度慢佔用空間大

min_after_dequeue=10*BATCH_SIZE

# the maximum number of elements in the queue佇列中的最大的資料量，建議
# 值為min_after_dequeue + (num_threads + a small safety margin) * batch_size

capacity=20*BATCH_SIZE

# shuffle the data to generate BATCH_SIZE sample pairs
data_batch,label_batch=tf.train.shuffle_batch([features,label],batch_size=BATCH_SIZE,
        capacity=capacity,min_after_dequeue=min_after_dequeue)

這樣就能使用data_batch和label_batch進行訓練了。
　　其實還有一種讀取方式速度更快一點，就是使用read_up_to的方式，這個在後面的補充部分會提到。

3、資料讀取

例1、前面已經提到

例2、

　　這裡首先再增加一個讀取二進位制檔案的例子。而這裡說的二進位制檔案是TensorFlow自己的二進位制檔案格式TFRecords。TFRecords是一個序列化的tf.train.Example Protobuf物件，可以用幾行程式碼就將一個影象轉為一個二進位制文件。大致流程是首先讀取資料並把資料填充到一個Example protocol buffer中，然後序列化protocol buffer為一個string，並用tf.python_io.TFRecordWriter將這個string寫入TFRecords 中。
　　這裡用一個例子2來介紹：

#例2
#首先讀取影象並轉為byte字串
def get_image_binary (filename):
    image=Image.open(filename)
    image=np.asarray(image,np.uint8)
    shape=np.array(image.shape,np.int32)#儲存形狀資訊幫助復原
    # convert image to raw data bytes in the array.
    # 轉換影象為byte
    return shape.tobytes(),image.tobytes() 

為了把上面生成的byte strings寫入TFRecords需要使用tf.python_io.TFRecordWriter和tf.train.Features。

#例2
def write_to_tfrecord(label,shape,binary_image, tfrecord_file):
""" This example is to write a sample to TFRecord file. If you want to write more samples , just use a loop.
這個例子只是寫入一個取樣，想寫入多組資料只需要用個迴圈
"""
    # 定義寫入方法
    writer=tf.python_io.TFRecordWriter(tfrecord_file)
    # write label, shape, and image content to the TFRecord file
    # 正如所說的把資料填充到一個Example protocol buffer
    example=tf.train.Example(features=tf.train.Features(feature ={
        'label':tf.train.Feature(bytes_list=tf.train.BytesList(value=[label])),
        'shape':tf.train.Feature(bytes_list=tf.train.BytesList(value=[shape])),
        'image':tf.train.Feature(bytes_list=tf.train.BytesList(value=[binary_image]))
        }))
    #然後序列化protocol buffer為一個string，並寫入
    writer.write(example.SerializeToString())
    writer.close()

　　這樣就已經將資料寫入了TFRecord中了，當要讀取資料的時候，需要TFRecordReader 和tf.parse_single_example，但是注意這裡讀取的資料是一個tensor，想得到值還需要eval。

#例2
def read_from_tfrecord(filenames):
    tfrecord_file_queue=tf.train.string_input_producer(filenames, name='queue')
    reader=tf.TFRecordReader()
    _ , tfrecord_serialized=reader.read(tfrecord_file_queue )
    # label and image are stored as bytes but could be stored as
    # int64 or float64 values in a serialized tf.Example protobuf.
    tfrecord_features=tf.parse_single_example(tfrecord_serialized,
            features ={
                'label' : tf.FixedLenFeature([],tf.string ),
                'shape' : tf.FixedLenFeature([],tf.string ),
                'image' : tf.FixedLenFeature([],tf.string ),
                }, name='features')
    # image was saved as uint8, so we have to decode as uint8.
    image = tf.decode_raw(tfrecord_features['image'], tf.uint8 )
    shape = tf.decode_raw(tfrecord_features['shape'], tf.int32 )
    # the image tensor is flattened out, so we have to reconstruct the shape
    image=tf.reshape(image,shape)
    label=tf.cast(tfrecord_features['label'],tf.string)#型別轉換，因為只有一個值
    return label, shape, image

例3

程式和講解在我的github上。

4.進階補充

這裡對上面用到的一些函式進行更詳細的解釋：
　　我們線上程和佇列部分知道了需要一個QueueRunner生成一些執行緒反覆的執行enqueue操作，並且顯示的定義了一個queue和QueueRunner進行了一些操作，但是後面我們就直接用到了Reader和start_queue_runners這些內容有什麼關係呢　　

1. tf.train.start_queue_runners

　　用於啟動圖中所有的queue runners，我們的圖中包含兩個queue runners一個是檔名佇列，一個是記憶體中的資料的佇列。其建構函式為：

start_queue_runners(
    sess=None,這個是執行佇列操作Op的對話，預設是default session
    coord=None,可選：協調器用於協調起始執行緒
    daemon=True,是否執行緒標記為守護執行緒
    start=True,如果是FALSE就只創造執行緒而不啟動他們
    collection=tf.GraphKeys.QUEUE_RUNNERS#指定queue runners的來源預設是GraphKeys.QUEUE_RUNNERS
)

2.tf.ReaderBase

　　這個是所有Reader的基類，只要這個懂了就差不多都明白了。這個類簡單的說是提供了將string（通常為filename）內容提取出來的功能，每次只能提取一條內容，但是一個檔案有很多條內容而且我們又有很多的檔案存在，所以為了更方便需要用一個佇列來進行，佇列中儲存檔名然後在需要讀取的時候用Reader出隊。

3. batch用到的函式:

　　在batch的時候遇到了tf.train.batch等函式，這裡對這些函式的原理進行分析。這些函式在圖中加入了一個queue用於聚合a batch of examples，它們還加了一個QueueRunner用於填充佇列。

• 其中tf.train.batch和tf.train.shuffle_batch是單執行緒的產生資料，或是在有一個單一子圖用於生成資料的時候用N個執行緒來執行（N是能夠保證佇列滿的數量）
• tf.train.batch_join和tf.train.shuffle_batch_join用於多子圖多執行緒的生成資料。

這些看上去不好分辨，看一下建構函式就明白了。
　首先是tf.train.batch的建構函式：

batch(
    tensors,可以是列表或字典，這個函式的返回值和這個tensor中的值是一樣
    batch_size,
    num_threads=1,入隊所需的執行緒數
    capacity=32,佇列中最大的element數量
    enqueue_many=False,這個表示上面輸入的tensor是不是隻是一個example
    shapes=None,
    dynamic_pad=False,是否對於不同shape的例子進行補充
    allow_smaller_final_batch=False,
    shared_name=None,
    name=None
)

接著是tf.train.batch_join：

batch_join(
    tensors_list,這個是tensor的元組或字典列表，列表中的一個元素和上面batch中的tensors是一樣的，而啟動的執行緒數量等於len(tensors_list)
    batch_size,
    capacity=32,
    enqueue_many=False,
    shapes=None,
    dynamic_pad=False,
    allow_smaller_final_batch=False,
    shared_name=None,
    name=None
)

4.一種更高效的資料讀取方法read_up_to

　　我們前面的模式都是Reader.read+train_batch的形式，但是這樣還是和理想速度差很多。其實我們可以用read_up_to的方式提高讀取效率。這是Reader中的一個方法，其建構函式為：

read_up_to(
    queue,檔案佇列
    num_records,讀取資料的數量
    name=None
)
返回值是一個Tensor(keys,values)元組

其大概使用流程：

構造檔案佇列->構造reader物件->讀取n條資料->返回資料

而read+train_batch流程為:

構造檔案佇列->構造reader物件->讀取1條資料->將改資料加入佇列->若資料佇列長度大於n，從佇列中返回結果