如何使用tfrecord

Posted 2023-04-04

参考技术A 你好，基本使用

使用 TensorFlow, 你必须明白 TensorFlow:

使用图 (graph) 来表示计算任务.
在被称之为 会话 (Session) 的上下文 (context) 中执行图.
使用 tensor 表示数据.
通过 变量 (Variable) 维护状态.
使用 feed 和 fetch 可以为任意的操作(arbitrary operation) 赋值或者从其中获取数据.

综述

TensorFlow 是一个编程系统, 使用图来表示计算任务. 图中的节点被称之为 op
(operation 的缩写). 一个 op 获得 0 个或多个 Tensor, 执行计算,
产生 0 个或多个 Tensor. 每个 Tensor 是一个类型化的多维数组.
例如, 你可以将一小组图像集表示为一个四维浮点数数组,
这四个维度分别是 [batch, height, width, channels].

一个 TensorFlow 图描述了计算的过程. 为了进行计算, 图必须在 会话 里被启动.
会话 将图的 op 分发到诸如 CPU 或 GPU 之类的 设备 上, 同时提供执行 op 的方法.
这些方法执行后, 将产生的 tensor 返回. 在 Python 语言中, 返回的 tensor 是
numpy ndarray 对象; 在 C 和 C++ 语言中, 返回的 tensor 是
tensorflow::Tensor 实例.

计算图

TensorFlow 程序通常被组织成一个构建阶段和一个执行阶段. 在构建阶段, op 的执行步骤
被描述成一个图. 在执行阶段, 使用会话执行执行图中的 op.

例如, 通常在构建阶段创建一个图来表示和训练神经网络, 然后在执行阶段反复执行图中的训练 op.

TensorFlow 支持 C, C++, Python 编程语言. 目前, TensorFlow 的 Python 库更加易用,
它提供了大量的辅助函数来简化构建图的工作, 这些函数尚未被 C 和 C++ 库支持.

三种语言的会话库 (session libraries) 是一致的.

构建图

构建图的第一步, 是创建源 op (source op). 源 op 不需要任何输入, 例如 常量 (Constant). 源 op 的输出被传递给其它 op 做运算.

Python 库中, op 构造器的返回值代表被构造出的 op 的输出, 这些返回值可以传递给其它
op 构造器作为输入.

TensorFlow Python 库有一个默认图 (default graph), op 构造器可以为其增加节点. 这个默认图对
许多程序来说已经足够用了. 阅读 Graph 类 文档
来了解如何管理多个图.
import tensorflow as tf

# 创建一个常量 op, 产生一个 1x2 矩阵. 这个 op 被作为一个节点
# 加到默认图中.
#
# 构造器的返回值代表该常量 op 的返回值.
matrix1 = tf.constant([[3., 3.]])

# 创建另外一个常量 op, 产生一个 2x1 矩阵.
matrix2 = tf.constant([[2.],[2.]])

# 创建一个矩阵乘法 matmul op , 把 'matrix1' 和 'matrix2' 作为输入.
# 返回值 'product' 代表矩阵乘法的结果.
product = tf.matmul(matrix1, matrix2)

默认图现在有三个节点, 两个 constant() op, 和一个matmul() op. 为了真正进行矩阵相乘运算, 并得到矩阵乘法的
结果, 你必须在会话里启动这个图.

在一个会话中启动图

构造阶段完成后, 才能启动图. 启动图的第一步是创建一个 Session 对象, 如果无任何创建参数,
会话构造器将启动默认图.

欲了解完整的会话 API, 请阅读Session 类.
# 启动默认图.
sess = tf.Session()

# 调用 sess 的 'run()' 方法来执行矩阵乘法 op, 传入 'product' 作为该方法的参数.
# 上面提到, 'product' 代表了矩阵乘法 op 的输出, 传入它是向方法表明, 我们希望取回
# 矩阵乘法 op 的输出.
#
# 整个执行过程是自动化的, 会话负责传递 op 所需的全部输入. op 通常是并发执行的.
#
# 函数调用 'run(product)' 触发了图中三个 op (两个常量 op 和一个矩阵乘法 op) 的执行.
#
# 返回值 'result' 是一个 numpy `ndarray` 对象.
result = sess.run(product)
print result
# ==> [[ 12.]]

# 任务完成, 关闭会话.
sess.close()

Session 对象在使用完后需要关闭以释放资源. 除了显式调用 close 外, 也可以使用 "with" 代码块
来自动完成关闭动作.
with tf.Session() as sess:
result = sess.run([product])
print result

在实现上, TensorFlow 将图形定义转换成分布式执行的操作, 以充分利用可用的计算资源(如 CPU
或 GPU). 一般你不需要显式指定使用 CPU 还是 GPU, TensorFlow 能自动检测. 如果检测到 GPU, TensorFlow
会尽可能地利用找到的第一个 GPU 来执行操作.

如果机器上有超过一个可用的 GPU, 除第一个外的其它 GPU 默认是不参与计算的. 为了让 TensorFlow
使用这些 GPU, 你必须将 op 明确指派给它们执行. withDevice 语句用来指派特定的 CPU 或 GPU
执行操作:
with tf.Session() as sess:
with tf.device("/gpu:1"):
matrix1 = tf.constant([[3., 3.]])
matrix2 = tf.constant([[2.],[2.]])
product = tf.matmul(matrix1, matrix2)

设备用字符串进行标识. 目前支持的设备包括:

"/cpu:0": 机器的 CPU.
"/gpu:0": 机器的第一个 GPU, 如果有的话.
"/gpu:1": 机器的第二个 GPU, 以此类推.

阅读使用GPU章节, 了解 TensorFlow GPU 使用的更多信息.

交互式使用

文档中的 Python 示例使用一个会话 Session 来
启动图, 并调用 Session.run() 方法执行操作.

为了便于使用诸如 IPython 之类的 Python 交互环境, 可以使用
InteractiveSession 代替
Session 类, 使用 Tensor.eval()
和 Operation.run() 方法代替
Session.run(). 这样可以避免使用一个变量来持有会话.
# 进入一个交互式 TensorFlow 会话.
import tensorflow as tf
sess = tf.InteractiveSession()

x = tf.Variable([1.0, 2.0])
a = tf.constant([3.0, 3.0])

# 使用初始化器 initializer op 的 run() 方法初始化 'x'
x.initializer.run()

# 增加一个减法 sub op, 从 'x' 减去 'a'. 运行减法 op, 输出结果
sub = tf.sub(x, a)
print sub.eval()
# ==> [-2. -1.]

Tensor

TensorFlow 程序使用 tensor 数据结构来代表所有的数据, 计算图中, 操作间传递的数据都是 tensor.
你可以把 TensorFlow tensor 看作是一个 n 维的数组或列表. 一个 tensor 包含一个静态类型 rank, 和
一个 shape. 想了解 TensorFlow 是如何处理这些概念的, 参见
Rank, Shape, 和 Type.

变量

Variables for more details.
变量维护图执行过程中的状态信息. 下面的例子演示了如何使用变量实现一个简单的计数器. 参见
变量 章节了解更多细节.
# 创建一个变量, 初始化为标量 0.
state = tf.Variable(0, name="counter")

# 创建一个 op, 其作用是使 state 增加 1

one = tf.constant(1)
new_value = tf.add(state, one)
update = tf.assign(state, new_value)

# 启动图后, 变量必须先经过`初始化` (init) op 初始化,
# 首先必须增加一个`初始化` op 到图中.
init_op = tf.initialize_all_variables()

# 启动图, 运行 op
with tf.Session() as sess:
# 运行 'init' op
sess.run(init_op)
# 打印 'state' 的初始值
print sess.run(state)
# 运行 op, 更新 'state', 并打印 'state'
for _ in range(3):
sess.run(update)
print sess.run(state)

# 输出:

# 0
# 1
# 2
# 3

代码中 assign() 操作是图所描绘的表达式的一部分, 正如 add() 操作一样. 所以在调用 run()
执行表达式之前, 它并不会真正执行赋值操作.

通常会将一个统计模型中的参数表示为一组变量. 例如, 你可以将一个神经网络的权重作为某个变量存储在一个 tensor 中.
在训练过程中, 通过重复运行训练图, 更新这个 tensor.

Fetch

为了取回操作的输出内容, 可以在使用 Session 对象的 run() 调用 执行图时, 传入一些 tensor,
这些 tensor 会帮助你取回结果. 在之前的例子里, 我们只取回了单个节点 state, 但是你也可以取回多个
tensor:
input1 = tf.constant(3.0)
input2 = tf.constant(2.0)
input3 = tf.constant(5.0)
intermed = tf.add(input2, input3)
mul = tf.mul(input1, intermed)

with tf.Session() as sess:
result = sess.run([mul, intermed])
print result

# 输出:
# [array([ 21.], dtype=float32), array([ 7.], dtype=float32)]

需要获取的多个 tensor 值，在 op 的一次运行中一起获得（而不是逐个去获取 tensor）。

Feed

上述示例在计算图中引入了 tensor, 以常量或变量的形式存储. TensorFlow 还提供了 feed 机制, 该机制
可以临时替代图中的任意操作中的 tensor 可以对图中任何操作提交补丁, 直接插入一个 tensor.

feed 使用一个 tensor 值临时替换一个操作的输出结果. 你可以提供 feed 数据作为 run() 调用的参数.
feed 只在调用它的方法内有效, 方法结束, feed 就会消失. 最常见的用例是将某些特殊的操作指定为 "feed" 操作,
标记的方法是使用 tf.placeholder() 为这些操作创建占位符.

input1 = tf.placeholder(tf.float32)
input2 = tf.placeholder(tf.float32)
output = tf.mul(input1, input2)

with tf.Session() as sess:
print sess.run([output], feed_dict=input1:[7.], input2:[2.])

# 输出:
# [array([ 14.], dtype=float32)]

for a larger-scale example of feeds.
如果没有正确提供 feed, placeholder() 操作将会产生错误.
MNIST 全连通 feed 教程
(source code)
给出了一个更大规模的使用 feed 的例子.

TensorFlow TFRecords简介

TensorFlow TFRecords简介

这篇博客将介绍TensorFlow的TFRecords，提供有关TFRecords的所有信息的一应俱全的介绍。从如何构建基本TFRecords到用于训练 SRGAN 和 ESRGAN 模型的高级TFRecords的所有内容。包括什么是TFRecords，如何序列化，反序列化数据，以及如何使用TFRecords预处理和序列化像div2k这样的大型数据集，如何使用TFRecords及TensorFlow训练深度神经网络。

TFRecord格式的两个主要优点是，高效地存储数据集，并且与从磁盘读取原始数据相比，获得了更快的I/O速度。

当使用TPU训练深度神经网络时，TFRecords非常有用。可以查看SRGAN和ESRGAN教程，其中介绍了如何使用Tensor处理单元（TPUs ensor Processing Units）和图形处理单元（GPUs Graphics Processing Units ）训练深度神经网络。

最好不使用tf.image.resize，坑太多

1. 效果图

可以看到原始数据和编码后数据相同，编码数据只是原始数据的字节字符串，TFRecord中的数据是序列化的二进制记录。

$ python single_tf_record.py

Original data: 12345
Encoded data: b'12345'
Data from the TFRecord: b'\\x05\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\xea\\xb2\\x04>12345z\\x1c\\xed\\xe8'
Decoded data: 12345

从输出中可以明显看出，原始数据被序列化为一系列字节字符串，随后被反序列化为原始数据。

$ python serialization.py

Original Data: [1 2 3 4]
Encoded Data: b'\\x08\\x04\\x12\\x04\\x12\\x02\\x08\\x04"\\x04\\x01\\x02\\x03\\x04'
Decoded Data: [1 2 3 4]

根据url下载网络图片，指定文件名，构建为TFRecord 数据，并序列化为二进制字符串保存到文件，然后读取在解析会照片和文件名，效果图如下：

2. 原理

2.1 安装

pip install tensorflow==2.1.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
# pip install tensorflow --upgrade -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install tensorflow-datasets

2.2 TFRecord是什么

TFRecord是用于存储二进制记录序列的自定义TensorFlow格式。TFRecords针对TensorFlow进行了高度优化，因此具有以下优势：

• 高效的数据存储形式
• 与其他类型的格式相比，读取速度更快

TFRecords最重要的用例之一是使用TPU训练模型。TPU功能强大，但需要远程存储与之交互的数据。在TPU上训练模型时，以TFRecord格式远程存储数据集，因为它可以有效地保存数据并更容易地加载数据。

2.3 什么是序列化二进制记录？

TFRecords存储一系列二进制记录。因此首先需要学习如何将数据转换为二进制表示。
TensorFlow有两个公共API，负责将数据编码和解码为二进制记录。这两个公共API来自tf.io.serialize_tensor 和 tf.io.parse_tensor。

通过使用tf.train.Feature进行数据的序列化和反序列化，支持的类型如下：

2.4 DIV2K数据集

DIVerse 2K分辨率高质量图像

• 1000张2K分辨率的图像分为：800张用于训练的图像、100张用于验证的图像和100张用于测试的图像
• 对于每个挑战赛道（具有1.双三次或2.未知降级运算符），
• 高分辨率图像：0001.png，0002.png，…，1000.png
• 缩小的图像：YYYYx2.png表示缩小因子x2；其中YYYY是图像ID;
  YYYYx3.png，缩小因子x3；
  YYYYx4.png；缩小因子x4
• DIV2K forder结构如下：
  DIV2K/–DIV2K数据集
  DIV2K/DIV2K_train_HR/–0001.png，0002.png，…，0800.png列车HR图像（提供给参与者）
  DIV2K/DIV2K_train_LR_bicubic/——使用Matlab调大小函数获得的具有默认设置的相应低分辨率图像（双三次插值）

3. 源代码

3.1 example_tf_record.py

# utils.py 从磁盘加载和保存图像到磁盘
# config.py 单个数据tfrecord示例的配置文件
# advance_config.py div2k数据集示例的配置文件
# single_tf_record.py 处理单个二进制记录并显示如何将其保存为TFRecord格式的脚本
# serialization.py 解释数据序列化重要性的脚本
# example_tf_record.py 保存和加载单个图片为TFRecord，如何从磁盘加载原始图像并以TFRecord格式对其进行序列化，以及如何加载序列化的TFRecord并对图像进行反序列化。
# create_tfrecords.py 生成高级TFRecords,保存和加载整个div2k数据集为TFRecords。将使用tfds（表示tensorflow_datasets，一组现成数据集）加载div2k数据集，对其进行预处理，然后将预处理的数据集序列化为TFRecords。
# DIV2K数据集：DIVerse 2K分辨率高质量图像
# 1000张2K分辨率的图像分为：800张用于训练的图像、100张用于验证的图像和100张用于测试的图像
# 对于每个挑战赛道（具有1.双三次或2.未知降级运算符），
# 高分辨率图像：0001.png，0002.png，…，1000.png
# 缩小的图像：YYYYx2.png表示缩小因子x2；其中YYYY是图像ID;
#           YYYYx3.png，缩小因子x3；
#           YYYYx4.png；缩小因子x4
# DIV2K forder结构如下：
# DIV2K/--DIV2K数据集
# DIV2K/DIV2K_train_HR/--0001.png，0002.png，…，0800.png列车HR图像（提供给参与者）
# DIV2K/DIV2K_train_LR_bicubic/——使用Matlab调大小函数获得的具有默认设置的相应低分辨率图像（双三次插值）

# USAGE
# python example_tf_record.py

import os

# 导入必要的包
import tensorflow as tf

from tfrecords_demo import config
from tfrecords_demo import utils

# 结构化的数据示例包括图片和图片名
# 从特定的url下载图像并将图像保存到磁盘。
imagePath = tf.keras.utils.get_file(
    config.IMAGE_FNAME,
    config.IMAGE_URL,
)

# 使用load_image函数从磁盘加载图像作为tf.Tensor
image = utils.load_image(pathToImage=imagePath)
class_name = config.IMAGE_CLASS

# 检查输出文件夹是否存在，不存在则创建
if not os.path.exists(config.OUTPUT_PATH):
    os.makedirs(config.OUTPUT_PATH)

# 保存缩放后的照片
utils.save_image(image=image, saveImagePath=config.RESIZED_IMAGE_PATH)

# 构建图片tf.train.Feature和类名tf.train.Feature
imageFeature = tf.train.Feature(
    bytes_list=tf.train.BytesList(value=[
        # 注意序列化图像的方法
        tf.io.serialize_tensor(image).numpy(),
    ])
)
classNameFeature = tf.train.Feature(
    bytes_list=tf.train.BytesList(value=[
        class_name.encode(),
    ])
)

# 包装图片和类名feature到一个feature字典中，并将其作为参数初始化一个类
features = tf.train.Features(feature=
    "image": imageFeature,
    "class_name": classNameFeature,
)
example = tf.train.Example(features=features)

# 序列化整个实例 使用SerializeToString函数直接序列化
serialized = example.SerializeToString()

# 将序列化实例写入 TFRecord
with tf.io.TFRecordWriter(config.TFRECORD_EXAMPLE_FNAME) as recordWriter:
    recordWriter.write(serialized)

# 构建feature模式和 TFRecord数据
featureSchema = 
    "image": tf.io.FixedLenFeature([], dtype=tf.string),
    "class_name": tf.io.FixedLenFeature([], dtype=tf.string),

# 读取数据构建TFRecord
dataset = tf.data.TFRecordDataset(config.TFRECORD_EXAMPLE_FNAME)

# 遍历数据
for element in dataset:
    # 获取序列化实例数据，并根据feature模式解析
    # 注意如何使用这里的特征示意图来解析示例。(序列化和反序列化时的数据类型是一样的）
    element = tf.io.parse_single_example(element, featureSchema)

    # 获取序列化后的类名和图像
    className = element["class_name"].numpy().decode()
    image = tf.io.parse_tensor(
        element["image"].numpy(),
        out_type=tf.dtypes.float32
    )

    # 使用图片名和图片保存反序列化后的图像
    utils.save_image(
        image=image,
        saveImagePath=config.DESERIALIZED_IMAGE_PATH,
        title=className
    )

3.2 create_tfrecords.py

# USAGE
# python create_tfrecords.py

# 导入必要的包
import os

import tensorflow as tf
import tensorflow_datasets as tfds

from tfrecords_demo import config

# 定义自动调频对象以优化过程
AUTO = tf.data.experimental.AUTOTUNE


def pre_process(element):
    # 获取低、高分辨率图像
    lrImage = element["lr"]
    hrImage = element["hr"]

    # 将低高分辨率图像从Tensor张量转换为序列化的张量TensorProto proto
    lrByte = tf.io.serialize_tensor(lrImage)
    hrByte = tf.io.serialize_tensor(hrImage)

    # 返回低、高分辨率proto对象
    return (lrByte, hrByte)


def create_dataset(dataDir, split, shardSize):
    print(config.DATASET, dataDir, shardSize)
    # 加载数据集，保存到磁盘，并处理
    ds = tfds.load(name="div2k", split=split, data_dir=dataDir,download=True)
    ds = (ds
          .map(pre_process, num_parallel_calls=AUTO)
          .batch(shardSize)
          )

    # 返回数据集TensorFlow dataset object
    return ds


def create_serialized_example(lrByte, hrByte):
    # 创建低、高分辨率图像字节list
    lrBytesList = tf.train.BytesList(value=[lrByte])
    hrBytesList = tf.train.BytesList(value=[hrByte])

    # 从字节list构建低、高分辨率推向feature
    lrFeature = tf.train.Feature(bytes_list=lrBytesList)
    hrFeature = tf.train.Feature(bytes_list=hrBytesList)

    # 构建低、高分辨率图像feature字典
    featureMap = 
        "lr": lrFeature,
        "hr": hrFeature,
    

    # 构建一个features集合，构建features实例，序列化实例
    features = tf.train.Features(feature=featureMap)
    example = tf.train.Example(features=features)
    serializedExample = example.SerializeToString()

    # 返回序列化的实例
    return serializedExample


def prepare_tfrecords(dataset, outputDir, name, printEvery=50):
    # 检查输出路径是否存在
    if not os.path.exists(outputDir):
        os.makedirs(outputDir)

    # 遍历数据集，创建 TFRecords
    for (index, images) in enumerate(dataset):
        # 获取分片数，构建名称
        shardSize = images[0].numpy().shape[0]
        tfrecName = f"index:02d-shardSize.tfrec"
        filename = outputDir + f"/name-" + tfrecName

        # 写入 tfrecords
        with tf.io.TFRecordWriter(filename) as outFile:
            # write shard size serialized examples to each TFRecord
            for i in range(shardSize):
                serializedExample = create_serialized_example(
                    images[0].numpy()[i], images[1].numpy()[i])
                outFile.write(serializedExample)

            # 打印进度
            if index % printEvery == 0:
                print("[INFO] wrote file  containing  records..."
                      .format(filename, shardSize))


# ds = tfds.load('mnist', split='train', shuffle_files=True)
# ds = tfds.load('div2k', split='train[:5%]', shuffle_files=True)

# 创建div2k images的训练和验证数据集
print("[INFO] creating div2k training and testing dataset...")
trainDs = create_dataset(dataDir=config.DIV2K_PATH, split="train[:5%]",
                         shardSize=config.SHARD_SIZE)
testDs = create_dataset(dataDir=config.DIV2K_PATH, split="validation",
                        shardSize=config.SHARD_SIZE)

# 创建训练和测试 TFRecords，并写入磁盘
print("[INFO] preparing and writing div2k TFRecords to disk...")
prepare_tfrecords(dataset=trainDs, name="train",
                  outputDir=config.GPU_DIV2K_TFR_TRAIN_PATH)
prepare_tfrecords(dataset=testDs, name="test",
                  outputDir=config.GPU_DIV2K_TFR_TEST_PATH)

4. 报错及解决

1. tf.data.experimental.AUTOTUNE
2. tensorflow >=2.1.0
   

参考

• https://pyimagesearch.com/2022/08/08/introduction-to-tfrecords/
• div2k数据集