一、程序介绍

1、包导入

# Author : Hellcat

# Time   : 17-12-29

import os

import numpy as np

np.set_printoptions(threshold=np.inf)

import tensorflow as tf

config = tf.ConfigProto()

config.gpu_options.allow_growth = True

sess = tf.Session(config=config)

from tensorflow.examples.tutorials.mnist import input_data

2、TFRecord录入格式转换

TFRecord的录入格式是确定的，整数或者二进制，在train函数中能查看所有可以接受类型

def _int64_feature(value):

    """生成整数数据属性"""

    return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))

def _bytes_feature(value):

    """生成字符型数据属性"""

    return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))

3、TFRecord文件写入测试

将mnist数据以每张图片为单位写入同一个TFR文件，

实际上就是每次把一个图片相关信息都写入，注意文件类型，二级制数据需要以string的格式保存

def TFRecord_write():

    """将mnist数据集写入TFR文件"""

    mnist = input_data.read_data_sets('./Data_Set/Mnist_data',

                                      dtype=tf.uint8,one_hot=True)

    images = mnist.train.images

    labels = mnist.train.labels

    pixels = images.shape[1]                     # 784

    num_examples = mnist.train.num_examples      # 55000

    # TFRecords文件地址

    filename = './TFRecord_Output/mnist_train.tfrecords'

    if not os.path.exists('./TFRecord_Output/'):

        os.makedirs('./TFRecord_Output/')

    # 创建一个writer书写文件

    writer = tf.python_io.TFRecordWriter(filename)

    for index in range(num_examples):

        # 提取单张图像矩阵并转换为字符串

        image_raw = images[index].tostring()

        # 将单张图片相关数据写入TFR文件

        example = tf.train.Example(features=tf.train.Features(feature={

            'pixels':  _int64_feature(pixels),

            'label':   _int64_feature(np.argmax(labels[index])),

            'img_raw': _bytes_feature(image_raw)

        }))

        writer.write(example.SerializeToString())  # 序列化为字符串

    writer.close()

调用，

if __name__=='__main__':

    TFRecord_write()

输出如下，

4、TFRecord文件读取测试

实际的读取基本单位和存入的基本单位是一一对应的，当然也可以复数读取，但是由于tf后续有batch拼接的函数，所以意义不大

def TFRecord_read():

    """从TFR文件读取mnist数据集合"""

    # 创建一个reader读取文件

    reader = tf.TFRecordReader()

    # 创建读取文件队列维护文件列表

    filename_queue = tf.train.string_input_producer(['./TFRecord_Output/mnist_train.tfrecords'])

    # 读取数据

    # 每次读取一个

    # _, serialized_example = reader.read(filename_queue)

    # 每次读取多个

    _, serialized_example = reader.read_up_to(filename_queue,10)

    # 解析样例

    # 解析函数选择必须和上面读取函数选择相一致

    # 解析单个样例

    # features = tf.parse_single_example(

    # 同时解析所有样例

    features = tf.parse_example(

        serialized_example,

        features={

            'img_raw': tf.FixedLenFeature([],tf.string),

            'pixels':    tf.FixedLenFeature([],tf.int64),

            'label':    tf.FixedLenFeature([],tf.int64),

        })

    # 解析二进制数据格式,将之按照uint8格式解析

    images = tf.decode_raw(features['img_raw'],tf.uint8)

    labels = tf.cast(features['label'],tf.int32)

    pixels = tf.cast(features['pixels'],tf.int32)

    batch_size = 2

    capacity = 1000 + 3 * batch_size

    images.set_shape([10,784])

    labels.set_shape(10)

    pixels.set_shape(10)

    image_batch, label_batch, pixel_batch = tf.train.batch(

        [images, labels, pixels], batch_size=batch_size, capacity=capacity)

    # 线程控制器

    coord = tf.train.Coordinator()

    threads = tf.train.start_queue_runners(sess=sess,coord=coord) # 这里指代的是读取数据的线程，如果不加的话队列一直挂起

    for i in range(10):

        # print(images, labels, pixels)

        # print(sess.run(images))

        image, label, pixel = sess.run([image_batch,label_batch,pixel_batch])

        # image, label, pixel = sess.run([images,labels,pixels])

        print(image.shape,label,pixel)

输出，

拼接batch尺寸为2，每次读取10个数据

可以看到，这里batch尺寸指定的实际上是读取次数

(2, 10, 784)

[[7 3 4 6 1 8 1 0 9 8]
 [0 3 1 2 7 0 2 9 6 0]]

[[784 784 784 784 784 784 784 784 784 784]
 [784 784 784 784 784 784 784 784 784 784]]
……

注意读取数目和解析数目选择的函数是要对应的，

# 读取数据

# 每次读取一个

# _, serialized_example = reader.read(filename_queue)

# 每次读取多个，这里指定10个

_, serialized_example = reader.read_up_to(filename_queue,10)

# 解析样例

# 解析函数选择必须和上面读取函数选择相一致

# 解析单个样例

# features = tf.parse_single_example()

# 同时解析所有样例

features = tf.parse_example()

值得注意的是这句，

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

虽然后续未必会调用（coord实际上还是会调用用于协调停止），但实际上控制着队列的数据读取部分的启动，注释掉后会导致队列有出无进进而挂起。

5、TFRecord文件批量生成

def TFR_gen():

    """TFR样例数据生成"""

    # 定义写多少个文件(数据量大时可以写入多个文件加速)

    num_shards = 2

    # 定义每个文件中放入多少数据

    instances_per_shard = 2

    for i in range(num_shards):

        file_name = './TFRecord_Output/data.tfrecords-{}-of-{}'.format(i,num_shards)

        writer = tf.python_io.TFRecordWriter(file_name)

        for j in range(instances_per_shard):

            example = tf.train.Example(features=tf.train.Features(feature={

                'i':_int64_feature(i),

                'j':_int64_feature(j),

                'list':_bytes_feature(bytes([1,2,3]))

            }))

            writer.write(example.SerializeToString())  # 序列化为字符串

        writer.close()

输出如下，

6、TFRecord文件读取测试

def TFR_load():

    """批量载入TFR数据"""

    # 匹配文件名

    files = tf.train.match_filenames_once('./TFRecord_Output/data.tfrecords-*')

    import glob

    # files = glob.glob('./TFRecord_Output/data.tfrecords-*')

    # 载入文件名

    filename_queue = tf.train.string_input_producer(files,shuffle=True)

    reader = tf.TFRecordReader()

    _,serialized_example = reader.read(filename_queue)

    features = tf.parse_single_example(

        serialized_example,

        features={

            'i':tf.FixedLenFeature([],tf.int64),

            'j':tf.FixedLenFeature([],tf.int64),

            'list':tf.FixedLenFeature([],tf.string)

        })

    '''

    # tf.train.match_filenames_once操作中产生了变量

    # 值得注意的是局部变量，需要用下面的初始化函数初始化

    sess.run(tf.local_variables_initializer())

    print(sess.run(files))

    coord = tf.train.Coordinator()

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

    for i in range(6):

        print(sess.run([features['i'],features['j']]))

    coord.request_stop()

    coord.join(threads)

    '''

    example, label, array = features['i'], features['j'], features['list']

    # 每个batch的中样例的个数

    batch_size = 3

    # 队列中样例的个数

    capacity = 1000 + 3 * batch_size

    suffer = False

    # batch操作实际代指的就是数据读取和预处理操作

    if suffer is not True:

        example_batch, label_batch, array_batch = tf.train.batch(

            [example, label, array], batch_size=batch_size, capacity=capacity)

    else:

        # 不同线程处理各自的文件

        # 随机包含各个线程选择文件名的随机和文件内部数据读取的随机

        example_batch, label_batch, array_batch = tf.train.shuffle_batch(

            [example, label, array], batch_size=batch_size, capacity=capacity,

            min_after_dequeue=30)

    sess.run(tf.local_variables_initializer())

    coord = tf.train.Coordinator()

    threads = tf.train.start_queue_runners(sess=sess, coord=coord)  # 这里指代的是读取数据的线程，如果不加的话队列一直挂起

    for i in range(2):

        cur_example_batch, cur_label_batch, cur_array_batch = sess.run([example_batch, label_batch, array_batch])

        print(cur_example_batch, cur_label_batch, cur_array_batch)

    coord.request_stop()

    coord.join(threads)

注意下面介绍，

# tf.train.match_filenames_once操作中产生了变量

# 值得注意的是局部变量，需要用下面的初始化函数初始化

sess.run(tf.local_variables_initializer())

batch生成的两个函数如下，

suffer = False

# batch操作实际代指的就是数据读取和预处理操作

if suffer is not True:

   example_batch, label_batch, array_batch = tf.train.batch(

       [example, label, array], batch_size=batch_size, capacity=capacity)

else:

    # 不同线程处理各自的文件

    # 随机包含各个线程选择文件名的随机和文件内部数据读取的随机

    example_batch, label_batch, array_batch = tf.train.shuffle_batch(

        [example, label, array], batch_size=batch_size, capacity=capacity,

        min_after_dequeue=30)
  • 单一文件多线程，那么选用tf.train.batch（需要打乱样本，有对应的tf.train.shuffle_batch）
  • 多线程多文件的情况，一般选用tf.train.batch_join来获取样本（打乱样本同样也有对应的tf.train.shuffle_batch_join使用）

二、batch和batch_join的说明

1、文件准备

$ 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

2、单个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()  #创建一个协调器，管理线程

    threads = tf.train.start_queue_runners(coord=coord)  #启动QueueRunner, 此时文件名队列已经进队。

    for i in range(10):

        print example.eval()   #取样本的时候，一个Reader先从文件名队列中取出文件名，读出数据，Decoder解析后进入样本队列。

    coord.request_stop()

    coord.join(threads)

# outpt

# Alpha1

# Alpha2

# Alpha3

# Bee1

# Bee2

# Bee3

# Sea1

# Sea2

# Sea3

# Alpha1

3、单个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解后数据会进入这个队列，再批量出队。

# 虽然这里只有一个Reader，但可以设置多线程，相应增加线程数会提高读取速度，但并不是线程越多越好。

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']

4、多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设置为2

# 使用tf.train.batch_join()，可以使用多个reader，并行读取数据。每个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'数是单个Reader读取，但是可以多线程。tf.train.batch_join'和tf.train.shuffle_batch_join可设置多Reader读取，每个Reader使用一个线程。至于两种方法的效率，单Reader时，2个线程就达到了速度的极限。多Reader时，2个Reader就达到了极限。所以并不是线程越多越快，甚至更多的线程反而会使效率下降。

在这个例子中， 虽然只使用了一个文件名队列， 但是TensorFlow依然能保证多个文件阅读器从同一次迭代(epoch)的不同文件中读取数据，知道这次迭代的所有文件都被开始读取为止。（通常来说一个线程来对文件名队列进行填充的效率是足够的）

另一种替代方案是： 使用tf.train.shuffle_batch 函数,设置num_threads的值大于1。 这种方案可以保证同一时刻只在一个文件中进行读取操作(但是读取速度依然优于单线程)，而不是之前的同时读取多个文件。这种方案的优点是：

  • 避免了两个不同的线程从同一个文件中读取同一个样本。
  • 避免了过多的磁盘搜索操作。

简单来说，

单一文件多线程，那么选用tf.train.batch（需要打乱样本，有对应的tf.train.shuffle_batch）

多线程多文件的情况，一般选用tf.train.batch_join来获取样本（打乱样本同样也有tf.train.shuffle_batch_join）

最新文章

  1. ubuntu下golang环境配置
  2. knockout源码分析之执行过程
  3. 用python来调试网络程序
  4. redis+Keepalived实现Redis主从复制
  5. 2016.5.27 PHP连接数据库与查询
  6. GLSL实现简单硬件Anisotrop Lighting 【转】
  7. oracle数据库导入导出命令！(转)
  8. Sicily-1050 深度优先搜索
  9. python学习1（小白记录）
  10. Javascript高级程序设计——Javascript简史+使用Javascript
  11. C#监控类属性的更改（大花猫动了哪些小玩具）
  12. bzoj 1196: [HNOI2006]公路修建问题
  13. 在虚拟机中安装metasploit官方攻防模拟器
  14. 认识多线程中start和run方法的区别?
  15. docker之harbor仓库注意事项
  16. Nginx + Tomcat 在 Windows7 上搭建负载均衡集群
  17. thinkpad的E480安装ubuntu后wifi无法使用问题解决
  18. c++ 出现“ error LNK2019: 无法解析的外部符号 该符号在函数 中被引用"错误原因
  19. 〖Android〗利用droidsshd在Android手机中开启 sshd,sftp,..
  20. 在不同的浏览器使用不同的css样式,解决浏览器兼容问题

热门文章

  1. TCP/IP的分层管理
  2. Session实现原理分析
  3. Nginx 解析PHP的原理 | CGI、FastCGI及php-fpm的关系
  4. mybites
  5. 泛型-----键值对----映射 hashmap--entry中key value 链表
  6. ACM总结——2017ACM-ICPC北京赛区现场赛总结
  7. flutter_webview_plugin 无法加载网页的异常处理
  8. Determine YARN and MapReduce Memory Configuration Settings
  9. Excel提取设定的多个关键字段
  10. ant_<target>标签含义与使用