我有一个大约有四千万行的CSV文件,每一行都是一个训练实例。根据TFRecords的文档,我试图对数据进行编码并保存在TFRecord文件中。
我找到的所有示例(甚至是TensorFlow仓库中的示例)都展示了创建TFRecord的过程取决于类TFRecordWriter。该类具有一个write
方法,它以序列化的字符串表示形式的数据作为输入,并将其写入磁盘。但是,这似乎是逐个训练实例完成的。
我该如何批量写入序列化数据?
假设我有一个函数:
def write_row(sentiment, text, encoded):
feature = {"one_hot": _float_feature(encoded),
"label": _int64_feature([sentiment]),
"text": _bytes_feature([text.encode()])}
example = tf.train.Example(features=tf.train.Features(feature=feature))
writer.write(example.SerializeToString())
将四千万次写入磁盘(每个样本一次)会非常缓慢。更有效的方法是将数据分批处理,每次写入50k或100k个样本(根据机器资源)。然而,在TFRecordWriter
内部似乎没有可用的批处理方法。
大致上可以这样实现:
class MyRecordWriter:
def __init__(self, writer):
self.records = []
self.counter = 0
self.writer = writer
def write_row_batched(self, sentiment, text, encoded):
feature = {"one_hot": _float_feature(encoded),
"label": _int64_feature([sentiment]),
"text": _bytes_feature([text.encode()])}
example = tf.train.Example(features=tf.train.Features(feature=feature))
self.records.append(example.SerializeToString())
self.counter += 1
if self.counter >= 10000:
self.writer.write(os.linesep.join(self.records))
self.counter = 0
self.records = []
但是当我读取使用该方法创建的文件时,我会收到以下错误:
tensorflow/core/framework/op_kernel.cc:1192] Invalid argument: Could not parse example input, value: '
��
label
��
one_hot����
��
注意:我可以更改编码过程,以便每个example
proto包含数千个示例而不仅仅是一个,但是我不想在写入TFrecord文件时以这种方式预先批处理数据,因为当我想要使用该文件进行具有不同批大小的训练时,它会在我的训练流水线中引入额外的开销。
with open("path", "wb")
方法提供了一个不需要额外费用的缓冲写入器。然而我找不到任何方法来检查TFRecordWriter
类是否在写入磁盘之前也进行了流缓冲... - Insectatoriousopen('...', 'b')
操作。 - bluesummersTFRecordWriter
。我添加了一个部分来说明TFRecordWriter
是带缓冲的。 - de1