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'\nR\n\x14\n\x08feature2\x12\x08\n\x06\.....# 反序列化 tf.train.Exampleexample_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_data

2.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': <tf.Tensor: shape=(), dtype=int64, numpy=0>, 'feature1': <tf.Tensor: shape=(), dtype=int64, numpy=4>, 'feature2': <tf.Tensor: shape=(), dtype=string, numpy=b'goat'>, 'feature3': <tf.Tensor: shape=(), dtype=float32, numpy=0.5251196>}

# 读取
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_listkind = 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