<jupyter_start><jupyter_text>tensorflow读取外部文件数据的过程：1. 文件系统——>文件名队列tf.train.string_input_producer(string_tensor,num_epochs=None,shuffle=True,seed=None,capactiy=32,shared_name=None,name=None,cancel_op=None)string_tensor:A 1-Dstring tensor with the strings to produceReturns: A queue with the output strings.A QueueRuner for the Queue is added to the current Graph's QUEUE_RUNNER collection.该函数创建了一个文件名队列，同时在图中当前图中添加了一个QUEUE_RUNNER用来管理读写该队列的线程，它是TF中对操作Queue的线程的封装。在队列被创建成功后，整个系统其实还是处于“停滞状态”，并没有文件名被加入到队列中，此时开始计算，会导致计算单元一直等待，导致整个系统被阻塞。使用tf.train.start_queue_runners之后，队列开始填充，系统不再停滞。也就是说我们在操作队列时，由于读入与读出之间的时间差，我们往往需要使用多线程进行处理，所以为了方便管理执行这些操作的线程，使用QueueRunner，在启动这些线程时，使用create_threads。2. 文件名队列——>内存队列<jupyter_code>import sys
# 定义一个队列
q = tf.FIFOQueue(10, "float") 
counter = tf.Variable(0.0)  #计数器
# 给计数器加一（这是一个操作）
increment_op = tf.assign_add(counter, 1.0)
# 将计数器加入队列（这是一个入队操作）
enqueue_op = q.enqueue(counter)

# 显示地创建QueueRunner，管理执行上面两个操作的线程
# 用多个线程向队列添加数据
# 这里实际创建了4个线程，两个增加计数，两个执行入队
qr = tf.train.QueueRunner(q, enqueue_ops=[increment_op, enqueue_op] * 2)

# 主线程
sess = tf.InteractiveSession()
tf.global_variables_initializer().run()
# 启动入队线程
qr.create_threads(sess, start=True)
for i in range(20):
    print (sess.run(q.dequeue()))<jupyter_output>520.0
534.0
570.0
580.0
590.0
723.0
734.0
747.0
763.0
778.0
785.0
795.0
806.0
821.0
838.0
853.0
871.0
889.0
901.0
918.0<jupyter_text>值得注意的是，上面的代码在执行过程中，执行计数的线程不断运行，执行入队的线程会先执行10次（因为队列为10），当一部分数据被消费后（这个出队的操作由主线程控制），入队线程继续执行。而当主线程的20个循环走完后，计数和入队的线程并不会停止（其中入队的线程在队列满10个时，处于阻塞状态，而计数不会停止） 为了避免上面情况的发生，即一些线程停止但另一些还在运行。tensorflow使用coordinator（协调器）协调线程组的运行状态。<jupyter_code>import tensorflow as tf
import threading, time

# 子线程函数
def loop(coord, id):
    t = 0
    while not coord.should_stop():
        print(id)
        time.sleep(2)
        t += 1
        # 只有1号线程调用request_stop方法
        if (t >= 2 and id == 1):
            coord.request_stop()

# 主线程
coord = tf.train.Coordinator()
# 使用Python API创建10个线程
threads = [threading.Thread(target=loop, args=(coord, i)) for i in range(10)]

# 启动所有线程，并等待线程结束
for t in threads: t.start()
coord.join(threads)<jupyter_output>0
1
2
3
4
5
67

8
9
0
1
2
3
4
5
6
7
98

0<jupyter_text>在上面的程序中，所有线程通过调用协调器的should_stop()方法来判断是否停止线程。可以发现当线程id=1执行到t=2时，将调用协调器的request_stop()方法，将should_stop()设置为true。则其他线程也都终止。主线程会等待所有子线程都停止后结束。 我们在前面知道，在操作队列时需要创建线程管理器QUENE_RUNNER，再通过create_threads方法启动线程，这是一种显示的进行队列操作的方式。<jupyter_code>import numpy as np
# 10个4维输入向量，每个数取值为1-10之间的随机数
data = 10 * np.random.randn(10, 4) + 1
# 1000个随机的目标值，值为0或1
target = np.random.randint(0, 2, size=10)

# 创建Queue，队列中每一项包含一个输入数据和相应的目标值
queue = tf.FIFOQueue(capacity=50, dtypes=[tf.float32, tf.int32], shapes=[[4], []])

# 批量入列数据（这是一个Operation）
enqueue_op = queue.enqueue_many([data, target])
# 出列数据（这是一个Tensor定义）
data_sample, label_sample = queue.dequeue()

# 创建包含4个线程的QueueRunner
qr = tf.train.QueueRunner(queue, [enqueue_op] * 4)

with tf.Session() as sess:
    # 创建Coordinator
    coord = tf.train.Coordinator()
    # 启动QueueRunner管理的线程
    enqueue_threads = qr.create_threads(sess, coord=coord, start=True)
    # 主线程，消费100个数据
    for step in range(100):
        if coord.should_stop():
            break
        data_batch, label_batch = sess.run([data_sample, label_sample])
        print([data_batch,label_batch])
    # 主线程计算完成，停止所有采集数据的进程 
    coord.request_stop() 
    #request that the threads stop,after this is called,calle to should_stop() will return True.
    coord.join(enqueue_threads)
    # Wait for threads to terminate.<jupyter_output>[array([  0.9855609,  -4.737697 , -20.064404 ,   6.8976574], dtype=float32), 1]
[array([ 4.241793 ,  6.925385 , -5.6965413,  5.1872787], dtype=float32), 1]
[array([ 15.496039 ,   5.0222635,  23.84017  , -10.769951 ], dtype=float32), 1]
[array([  8.540631,  -4.458069, -21.396536,  16.745102], dtype=float32), 1]
[array([ -2.627375 ,   5.0943184, -10.00617  ,  17.73765  ], dtype=float32), 1]
[array([-11.748537,  25.893698,  10.663585,  -8.087724], dtype=float32), 1]
[array([ 4.9214973,  0.9360481,  4.313497 , -2.1026533], dtype=float32), 0]
[array([ 9.659984 ,  7.596754 , -5.2508945,  8.074907 ], dtype=float32), 1]
[array([11.771369 ,  0.7063998,  2.0989969, -1.6823719], dtype=float32), 1]
[array([-0.9249712,  7.794566 ,  5.4953156, -6.9203296], dtype=float32), 1]
[array([  0.9855609,  -4.737697 , -20.064404 ,   6.8976574], dtype=float32), 1]
[array([ 4.241793 ,  6.925385 , -5.6965413,  5.1872787], dtype=float32), 1]
[array([ 15.496039 ,   5.0222635,  23.84017  , -10.769951 ], dtype=float[...]<jupyter_text>另一种启动线程的方式是采用start_queue_runners方法。在前面我们知道tf.train.string_input_producer在创建一个文件名队列时，将一个隐含的QueueRunner添加到全局图中。由于没有显示地返回QueueRunner来用create_threads启动线程，这里使用tf.train.start_queue_runners方法直接启动tf.GraphKeys.QUEUE_RUNNERS集合中的所有队列线程。<jupyter_code># 创建文件名队列filename_queue
filename_queue=tf.train.string_input_producer(['stat0.csv','stat1.csv'])  # 这两个文件是自定义的
# 创建读取CSV文件的TextLineReader
reader = tf.TextLineReader(skip_header_lines=1)
# 从文件名队列中取出CSV文件中的一条数据记录value
_,value=reader.read(filename_queue)
# 设置数据记录的默认值
record_defaults = [[1],[1],[1],[1]]
# 使用decode_csv方法将数据记录的每个字段都转换为特征张量
i_d,age,income,outgo=tf.decode_csv(value,record_defaults=record_defaults)
# 将所有特征张量组合在一起形成一条记录
features=tf.stack([i_d,age,income,outgo])
init_op = tf.group(tf.global_variables_initializer(),tf.local_variables_initializer())
with tf.Session() as sess:
    sess.run(init_op)
    coord = tf.train.Coordinator()
    # 启动计算图中所有的队列线程
    threads=tf.train.start_queue_runners(coord=coord)
    # 主线程，消费100个数据
    for _ in range(5):
        example=sess.run(features)
        print(example)
    coord.request_stop()
    coord.join(threads)<jupyter_output>[   1   24 2048 1024]
[   2   48 4096 2048]
[   1   24 2048 1024]
[   2   48 4096 2048]
[   2   48 4096 2048]<jupyter_text>关于TF中Example的定义：    message Example {      Features features = 1;    };    message Features {      // Map from feature name to feature.      map feature = 1;    };    message Feature {      // Each feature can be exactly one kind.      oneof kind {        BytesList bytes_list = 1;        FloatList float_list = 2;        Int64List int64_list = 3;      }    };    message BytesList {      repeated bytes value = 1;    }    message FloatList {      repeated float value = 1 [packed = true];    }    message Int64List {      repeated int64 value = 1 [packed = true];    } TFRecords文件存储是有结构的序列化字符块，它是Tensorflow推荐的标准文件格式。Tensorflow通过Protocol Buffers定义了TFRecords文件中存储的数据记录及其所含字段的数据结构，它们分别定义在example.proto和feature.proto文件中。一个样例包含一组特征，一组特征由多个特征向量由多个特征向量组成的Python字典构成。<jupyter_code># 创建向TFRecords文件写数据记录的writer
writer = tf.python_io.TFRecordWriter('stat.tfrecord')
# 两轮循环构造输入样例
for i in range(1,3):
    # 创建example.proto中定义的样例
    example = tf.train.Example(
        features=tf.train.Features(
            feature={
                'id':tf.train.Feature(int64_list=tf.train.Int64List(value=[i])),
                'age':tf.train.Feature(int64_list=tf.train.Int64List(value=[i*24])),
                'income':tf.train.Feature(float_list=tf.train.FloatList(value=[i*2048.0])),
                'outgo':tf.train.Feature(float_list=tf.train.FloatList(value=[i*1024.0]))
                }
        )
    )
    # 将样例序列化为字符串后，写入stat,tfrecord文件
    writer.write(example.SerializeToString())
# 关闭输入流
writer.close()
# tf.parse_single_example(serialized,features,name=None,example_names=None)
# serialized: A scalar string Tensor, a single serialized Example. 
# features: A dict mapping feature keys to FixedLenFeature or VarLenFeature values.
# Returns:A dict mapping feature keys to Tensor and SparseTensor values.

import tensorflow as tf
# 创建文件名队列filename_queue
filename_queue = tf.train.string_input_producer(['stat.tfrecord'],num_epochs=2)
# num_epochs规定遍历文件几次。
# 创建读取TFRecords文件的reader
reader = tf.TFRecordReader()
# 取出stat.tfrecord文件中的一条序列化的样例serialized_example
serialized_example = reader.read(filename_queue)
# 将一条序列化的样例转换为其包含的所有特征张量
# features = tf.parse_single_example(
# serialized_example,
# features={
#    'id':tf.FixedLenFeature([],tf.int64),
 #   'age':tf.FixedLenFeature([],tf.int64),
 #   'income':tf.FixedLenFeature([],tf.float32),
 #   'outgo':tf.FixedLenFeature([],tf.float32),
# })
serialized_example
init_op = tf.group(tf.global_variables_initializer(),tf.local_variables_initializer())
# 当使用coordiator管理线程组时，需要先执行tf.local_variables_initializer方法对其初始化
with tf.Session() as sess:
    # 初始化
    sess.run(init_op)
    coord =tf.train.Coordinator()
    threads = tf.train.start_queue_runners(sess=sess,coord=coord)
    # 打印程序的后台线程信息
    print('Thread:%s'%threads)
    try:
        for _ in range(10):   # 由于上文中num_epoches规定为2，也就是只能遍历文件2次，一个文件中包含2条记录，所以只能取出4条记录
            if not coord.should_stop():
                example=sess.run(features)
                print(example)
    except tf.errors.OutOfRangeError:
        print('Catch OutOfRangeError')
    finally:
        # 请求停止所有后台线程
        coord.request_stop()
        print('Finish reading')
    # 等待所有后台线程安全退出
    coord.join(threads)<jupyter_output>Thread:[<Thread(QueueRunnerThread-input_producer-input_producer/input_producer_EnqueueMany, started daemon 197652)>, <Thread(QueueRunnerThread-input_producer-close_on_stop, started daemon 214516)>, <Thread(QueueRunnerThread-input_producer_1-input_producer_1/input_producer_1_EnqueueMany, started daemon 214512)>, <Thread(QueueRunnerThread-input_producer_1-close_on_stop, started daemon 204932)>, <Thread(QueueRunnerThread-input_producer_2-input_producer_2/input_producer_2_EnqueueMany, started daemon 209800)>, <Thread(QueueRunnerThread-input_producer_2-close_on_stop, started daemon 191308)>, <Thread(QueueRunnerThread-input_producer_3-input_producer_3/input_producer_3_EnqueueMany, started daemon 213964)>, <Thread(QueueRunnerThread-input_producer_3-close_on_stop, started daemon 207548)>, <Thread(QueueRunnerThread-input_producer_4-input_producer_4/input_producer_4_EnqueueMany, started daemon 215088)>, <Thread(QueueRunnerThread-input_producer_4-close_on_stop, started daemon 207396)>, <Thread(Qu[...]
{'fraction_non_alphanumeric': 0.12191450933172787, 'fraction_numerical': 0.08037326911499097, 'mean_word_length': 5.03694730466384, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '209808', 'license': 'MIT', 'n_tokens_mistral': 4950, 'n_tokens_neox': 4815, 'n_words': 581, 'path': 'mlapp/MLAPP_CODE/MLAPP-C2-Code/.ipynb_checkpoints/Data Process In TF-checkpoint.ipynb', 'repo_name': 'xishansnow/MLAPP', 'size': '9966'}