1. tf.train.Example
1.1 tfrecord 数据范式转化1.2 demo 数据集构建 2. TFRecord 读写
2.1 写入1-tf.io.TFRecordWriter()2.3 读取-tf.data.TFRecordDataset()2.3 data -> dataset -> 存储-tf.data.experimental.TFRecordWriter()
tfrecord 用于存储二进制序列数据的一种范式，按顺序存，按顺序取。里面存的每一条数据都是一个 byte-string， 最常用的转byte-string的方式是tf.train.Example 。tf.train.Example (or protobuf) 以字典{“string”: value}的形式存储消息，这种消息存储机制可读性高。
demo1–tfrecord存储
value can be a num / list / array pybyte_value = np.array(value).tobytes() # 0.转Python字节数据 tfbyte_value = tf.train.BytesList(value=[pybyte_value]) # 1.转tf.train 字节数据 feature_dict[key] = tf.train.Feature(bytes_list=tfbyte_value)# 2.转tf.train.Feature(）注意是tf.train.Feature(）没有s .......... feature_example = tf.train.Example(features=tf.train.Features(feature=tffeature_dict)）# 3.转tf.train.Example() 注意tf.train.Features(）s exmp_serial = feature_example.SerializeToString() # 序列化feature_example tf_writer = tf.python_io.TFRecordWriter(tfrecord_path) # 构建tf写句柄 tf_writer.write(exmp_serial) # 写入tf文件 tf_writer.close() # 关闭句柄
np.array().tobytes()构造包含数组中原始数据的Python字节数据
1. tf.train.Example须将用户数据转化为tfrecord 约定的格式，才能使用tfrecord 格式存储数据。
1.1 tfrecord 数据范式转化1-> tfrecord支持写入三种格式的数据：string，int64，float32，分别通过tf.train.BytesList、tf.train.Int64List、tf.train.FloatList写入tf.train.Feature中。【就是说数据要写入tf.train.Feature前必须使用tf.train.BytesList，tf.train.Int64List，tf.train.FloatList必须使用强制类型转换】
# python 数据类型转tf.train.BytesList、tf.train.Int64List、tf.train.FloatList # tf.train.BytesList：string、byte # tf.train.FloatList：float (float32)、double (float64) # tf.train.Int64List ：bool、enum、int32、uint32、int64、uint64 # 强制类型转换 value = 1 value_ed = tf.train.Int64List(value=[value])
2-> tf.train.Feature 接受tf.train.BytesList、tf.train.Int64List、tf.train.FloatList 类型的数据。以下为scalar 转 tf.train.Feature 的快捷函数。 not scalar 的数据只需要用np.array().tobytes()/tf.io.serialize_tensor 转换成binary-strings，然后使用以下借口函数封装成 tf.train.Feature 即可。
# input : a scalar input
# output: tf.train.Feature
def _bytes_feature(value):
"""Returns a bytes_list from a string / byte."""
if isinstance(value, type(tf.constant(0))):
value = value.numpy() # BytesList won't unpack a string from an EagerTensor.
return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))
def _float_feature(value):
"""Returns a float_list from a float / double."""
return tf.train.Feature(float_list=tf.train.FloatList(value=[value]))
def _int64_feature(value):
"""Returns an int64_list from a bool / enum / int / uint."""
return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))
3->tf.train.Feature 构成特征字典 -> 特征字典 转 Features message -> Features message 转 tf.train.Example -> tf.train.Example 序列化后可以存入tfrecord 文件。【 Note that the tf.train.Example message is just a wrapper around the Features message:】
1.2 demo 数据集构建构建一个包含10000个观测数据的数据集，每条数据包含4个特征：[bool, label_index, lable_string, random_score]
n_observations = int(1e4) # The number of observations in the dataset. feature0 = np.random.choice([False, True], n_observations) #Boolean feature, encoded as False or True. feature1 = np.random.randint(0, 5, n_observations) # Integer feature, random from 0 to 4. strings = np.array([b'cat', b'dog', b'chicken', b'horse', b'goat']) # String feature. feature2 = strings[feature1] feature3 = np.random.randn(n_observations) # Float feature, from a standard normal distribution.
单个样本转tf.train.Feature-> tf.train.Features -> tf.train.Example()->SerializeToString() 接口函数
def serialize_example(feature0, feature1, feature2, feature3):
# Create a Feature dict : {key: tf.train.Feature}
feature = {
'feature0': _int64_feature(feature0),
'feature1': _int64_feature(feature1),
'feature2': _bytes_feature(feature2),
'feature3': _float_feature(feature3),
}
# Create a Features message and conver to tf.train.Example.
example_proto = tf.train.Example(features=tf.train.Features(feature=feature))
return example_proto.SerializeToString()
观测序列化[serialized_example ]和反序列化[tf.train.Example()]的结果
for i in range(n_observations):
f0, f1, f2, f3 = feature0[i], feature1[i], feature2[i], feature3[i]
# 序列化 tf.train.Example 消息
serialized_example = serialize_example(f0, f1, f2, f3) # b'nRnx14nx08feature2x12x08nx06.....
# 反序列化 tf.train.Example
example_proto = tf.train.Example.FromString(serialized_example)
'''
features {
feature {
key: "feature0"
value {
int64_list {
value: 0
}
}
}
feature {
key: "feature1"
value {
int64_list {
value: 4
}
}
}
feature {
key: "feature2"
value {
bytes_list {
value: "goat"
}
}
}
feature {
key: "feature3"
value {
float_list {
value: 0.9876000285148621
}
}
}
}
'''
2. TFRecord 读写
tfrecord 中每一条record按照下面的范式存储。tfrecord 文件中并非只能存tf.train.Example 序列化的结果，tf.train.Example 只是将字典序列化的一种方法。任何 byte-string都能够存入TFRecord file。
uint64 length uint32 masked_crc32_of_length byte data[length] uint32 masked_crc32_of_data2.1 写入1-tf.io.TFRecordWriter()
# Write the `tf.train.Example` observations to the file.
with tf.io.TFRecordWriter(filename) as writer: # 获取写入句柄
for i in range(n_observations):
example = serialize_example(feature0[i], feature1[i], feature2[i], feature3[i])
writer.write(example)
2.3 读取-tf.data.TFRecordDataset()
# 读取tfrecord文件， 获取序列化的样本
filenames = [filename]
raw_dataset = tf.data.TFRecordDataset(filenames) # tf.data.Dataset 对象
for raw_record in raw_dataset.take(10): # 读取前10 条
print(repr(raw_record)) # raw_record序列化的样本
# 序列化样本反序列化
# tf.data.Dataset 在图中执行，feature_description能够建立数据集shape和type的signature。
feature_description = {
'feature0': tf.io.FixedLenFeature([], tf.int64, default_value=0),
'feature1': tf.io.FixedLenFeature([], tf.int64, default_value=0),
'feature2': tf.io.FixedLenFeature([], tf.string, default_value=''),
'feature3': tf.io.FixedLenFeature([], tf.float32, default_value=0.0),
}
def _parse_function(example_proto):
# Parse the input `tf.train.Example` proto using the dictionary above.
# 一次只解析一条数据： use tf.parse example 可以一次解析一个batch的数据
return tf.io.parse_single_example(example_proto, feature_description)
# 利用tf.data.Dataset.map 函数将_parse_function 应用于数据集raw_dataset中的每一个元素
parsed_dataset = raw_dataset.map(_parse_function) # 可以用的数据
# {'feature0': , 'feature1': , 'feature2': , 'feature3': }
# 读取
filenames = [filename]
raw_dataset = tf.data.TFRecordDataset(filenames)
# tf.train.Example.ParseFromString反序列化 得到的是tf.train.Example features, 很难直接使用
for raw_record in raw_dataset.take(1):
example = tf.train.Example()
example.ParseFromString(raw_record.numpy())
# tf.train.Example features 转 dict of numpy array
result = {}
for key, feature in example.features.feature.items():
# The values are the Feature objects which contain a `kind` which contains:
# one of three fields: bytes_list, float_list, int64_list
kind = feature.WhichOneof('kind')
result[key] = np.array(getattr(feature, kind).value)
2.3 data -> dataset -> 存储-tf.data.experimental.TFRecordWriter()
from_tensor_slices 将data 转成dataset-> 序列化dataset 中的每一个元素-> 存入tf record 文件
features_dataset = tf.data.Dataset.from_tensor_slices((feature0, feature1, feature2, feature3))
for f0,f1,f2,f3 in features_dataset.take(1): # 逐个获取数据
print(f0, f1, f2, f3)
# tf.Tensor(False, shape=(), dtype=bool)，tf.Tensor(4, shape=(), dtype=int64)，tf.Tensor(b'goat', shape=(), dtype=string)，tf.Tensor(0.5251196235602504, shape=(), dtype=float64)
# 序列化方式1：tf.data.Dataset.map 映射数据集中的每一个元素
# 对于自定义的序列化操作函数serialize_example。为了使其成为TensorFlow graph 的节点，须使用 tf.py_function封装；之后再使用tf.data.Dataset.map 映射序列化数据集中的每一个元素。
def tf_serialize_example(f0,f1,f2,f3):
# (自定义函数，函数输入，函数输出)
tf_string = tf.py_function(serialize_example,(f0, f1, f2, f3),tf.string)
return tf.reshape(tf_string, ()) # The result is a scalar.
serialized_features_dataset = features_dataset.map(tf_serialize_example)
# 序列化方式2：tf.data.Dataset.from_generator(）映射数据集中的每一个元素
def generator():
for features in features_dataset:
yield serialize_example(*features)
serialized_features_dataset = tf.data.Dataset.from_generator(
generator, output_types=tf.string, output_shapes=())
整个序列化的数据集写入tfrecord.
# 整个写入tfrecord filename = 'test.tfrecord' writer = tf.data.experimental.TFRecordWriter(filename) # 与1.0 的接口有些不太一样 writer.write(serialized_features_dataset)
参考资料：TFRecord and tf.train.Example
