2.3 从文件读取数据

2.3 从文件读取数据

TensorFlow 数据集 (tf.data): 2.3 从文件读取数据

tf.data API 提供了一种强大且灵活的方式来构建数据输入管道，用于训练、评估和预测 TensorFlow 模型。 从文件读取数据是构建这些管道的一个关键步骤。 本节将深入探讨如何使用 tf.data 从各种文件格式读取数据，包括文本文件、CSV 文件、TFRecord 文件等。

2.3.1 从文本文件读取数据

从文本文件读取数据是最基本的情况。 每一行文本通常代表一个样本。

代码示例：

import tensorflow as tf
# 创建一个示例文本文件
with open("my_data.txt", "w") as f:
    f.write("1,2,3\n")
    f.write("4,5,6\n")
    f.write("7,8,9\n")
# 使用 TextLineDataset 读取文本文件
dataset = tf.data.TextLineDataset("my_data.txt")
# 打印数据集中的每个元素
for line in dataset:
    print(line.numpy())
# 定义一个解析函数，将每行文本解析为数值
def parse_line(line):
    fields = tf.strings.split(line, ",").to_tensor()  # 使用 to_tensor 确保形状一致
    nums = tf.strings.to_number(fields, out_type=tf.float32)
    return nums
# 使用 map 应用解析函数
dataset = dataset.map(parse_line)
# 打印解析后的数据集中的每个元素
for item in dataset:
    print(item.numpy())
# 清理文件
import os
os.remove("my_data.txt")
​

代码详解：

1. tf.data.TextLineDataset("my_data.txt"): 创建一个 TextLineDataset 对象，它会逐行读取指定的文件。

2. dataset.map(parse_line): 使用 map 转换操作，将 parse_line 函数应用于数据集中的每个元素（每一行文本）。

3. parse_line(line): 这是一个用户定义的函数，负责将每一行文本解析成所需的格式。 在这个例子中，它将逗号分隔的字符串分割成数值，并转换为 tf.float32 类型。

   • tf.strings.split(line, ","): 将字符串按逗号分割成字符串张量。

   • .to_tensor(): 将稀疏张量转换为密集张量，确保所有行的形状相同，这对于后续的批处理操作至关重要。如果数据行长度不一致，可以使用 to_tensor(default_value=...) 来填充缺失值。

   • tf.strings.to_number(fields, out_type=tf.float32): 将字符串张量转换为数值张量，指定数据类型为 tf.float32。

Graph Visualization (Mermaid):

2.3.2 从 CSV 文件读取数据

CSV (Comma Separated Values) 文件是一种常见的数据存储格式。 tf.data 提供了方便的方法来读取和解析 CSV 文件。

代码示例：

import tensorflow as tf
import pandas as pd
# 创建一个示例 CSV 文件
data = {'col1': [1, 4, 7], 'col2': [2, 5, 8], 'col3': [3, 6, 9]}
df = pd.DataFrame(data)
df.to_csv("my_data.csv", index=False)
# 定义 CSV 文件的列名
CSV_COLUMNS = ['col1', 'col2', 'col3']
# 定义默认值，用于处理缺失数据
DEFAULTS = [[0.0], [0.0], [0.0]]
# 解析 CSV 行的函数
def decode_csv(line):
    fields = tf.io.decode_csv(line, record_defaults=DEFAULTS)
    features = dict(zip(CSV_COLUMNS, fields))
    return features
# 创建数据集
dataset = tf.data.TextLineDataset("my_data.csv").skip(1)  # 跳过标题行
dataset = dataset.map(decode_csv)
# 打印数据集中的每个元素
for item in dataset.take(2): # 只打印前两个元素
    print({k: v.numpy() for k, v in item.items()})
# 清理文件
import os
os.remove("my_data.csv")
​

代码详解：

1. tf.io.decode_csv(line, record_defaults=DEFAULTS): 这是解析 CSV 数据的关键函数。

   • line: 要解析的 CSV 字符串。

   • record_defaults: 一个列表，包含了每一列的默认值。 如果 CSV 文件中缺少某个值，则使用相应的默认值填充。 这对于处理缺失数据非常重要。

2. dataset.skip(1): 跳过CSV文件的首行，通常是表头。

3. dict(zip(CSV_COLUMNS, fields)): 将解析后的字段与列名关联起来，创建一个字典，方便访问。

4. .take(2): 只从数据集中取前两个元素，用于展示目的。

Graph Visualization (Mermaid):

处理不同数据类型：

record_defaults 参数允许你指定每一列的数据类型。 例如，如果你的 CSV 文件包含字符串列，你可以将相应的默认值设置为 [""]。

使用 tf.data.experimental.make_csv_dataset (更高级的方法):

TensorFlow 提供了一个更高级的函数 tf.data.experimental.make_csv_dataset，它可以自动处理 CSV 文件的读取、解析和批处理。

import tensorflow as tf
import pandas as pd
# 创建一个示例 CSV 文件
data = {'col1': [1, 4, 7], 'col2': [2, 5, 8], 'col3': [3, 6, 9]}
df = pd.DataFrame(data)
df.to_csv("my_data.csv", index=False)
# 使用 make_csv_dataset
dataset = tf.data.experimental.make_csv_dataset(
    "my_data.csv",
    batch_size=3,  # 指定批处理大小
    header=True,  # 指示文件包含标题行
    num_epochs=1  # 指定迭代次数
)
# 打印数据集中的一个批次
for batch in dataset:
    for k, v in batch.items():
        print(f"{k}: {v.numpy()}")
    break
# 清理文件
import os
os.remove("my_data.csv")
​

make_csv_dataset 简化了 CSV 数据的读取和批处理过程，使其更加方便。

2.3.3 从 TFRecord 文件读取数据

TFRecord 是一种 TensorFlow 特定的二进制文件格式，用于存储序列化的数据。 它可以有效地存储大量数据，并优化 TensorFlow 的读取性能。

代码示例：

import tensorflow as tf
# 创建一个示例 TFRecord 文件
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]))
def serialize_example(feature0, feature1, feature2):
  """
  Creates a tf.train.Example message ready to be written to a file.
  """
  # Create a dictionary mapping the feature name to the feature proto definition.
  feature = {
      'feature0': _int64_feature(feature0),
      'feature1': _float_feature(feature1),
      'feature2': _bytes_feature(feature2),
  }
  # Create a Features message using tf.train.Example.
  example_proto = tf.train.Example(features=tf.train.Features(feature=feature))
  return example_proto.SerializeToString()
with tf.io.TFRecordWriter("my_data.tfrecord") as writer:
    example = serialize_example(10, 3.14, b"hello")
    writer.write(example)
    example = serialize_example(20, 6.28, b"world")
    writer.write(example)
# 定义特征描述符
feature_description = {
    'feature0': tf.io.FixedLenFeature([], tf.int64, default_value=0),
    'feature1': tf.io.FixedLenFeature([], tf.float32, default_value=0.0),
    'feature2': tf.io.FixedLenFeature([], tf.string, default_value=''),
}
# 解析 TFRecord 的函数
def _parse_function(example_proto):
  # Parse the input tf.train.Example proto using the dictionary above.
  return tf.io.parse_single_example(example_proto, feature_description)
# 创建数据集
dataset = tf.data.TFRecordDataset("my_data.tfrecord")
dataset = dataset.map(_parse_function)
# 打印数据集中的每个元素
for item in dataset:
    print({k: v.numpy() for k, v in item.items()})
# 清理文件
import os
os.remove("my_data.tfrecord")
​

代码详解：

1. tf.io.TFRecordWriter("my_data.tfrecord"): 创建一个 TFRecordWriter 对象，用于将数据写入 TFRecord 文件。

2. serialize_example(...): 这是一个用户定义的函数，负责将数据序列化为 tf.train.Example 协议缓冲区。 tf.train.Example 是 TFRecord 中存储数据的标准格式。

3. tf.io.parse_single_example(example_proto, feature_description): 这是解析 TFRecord 数据的关键函数。

   • example_proto: 要解析的 tf.train.Example 协议缓冲区。

   • feature_description: 一个字典，描述了每个特征的类型和形状。 这告诉 TensorFlow 如何解析数据。

4. tf.data.TFRecordDataset("my_data.tfrecord"): 创建一个 TFRecordDataset 对象，它会读取指定的 TFRecord 文件。

5. dataset.map(_parse_function): 使用 map 转换操作，将 _parse_function 函数应用于数据集中的每个元素（每个序列化的 tf.train.Example）。

Graph Visualization (Mermaid):

TFRecord 的优势：

• 高效的存储： TFRecord 是一种二进制格式，可以有效地存储大量数据。

• 优化的读取性能： TensorFlow 可以高效地读取 TFRecord 文件，减少了 I/O 开销。

• 数据类型支持： TFRecord 支持各种数据类型，包括数值、字符串和图像。

何时使用 TFRecord：

当处理大型数据集时，特别是图像数据或需要高性能 I/O 的场景，TFRecord 是一个不错的选择。

2.3.4 高级技巧

• 数据洗牌 (Shuffling): 使用 dataset.shuffle(buffer_size) 对数据进行洗牌，以避免模型训练中的偏差。 buffer_size 指定了洗牌缓冲区的大小。

• 批处理 (Batching): 使用 dataset.batch(batch_size) 将数据分成批次，以提高训练效率。

• 预取 (Prefetching): 使用 dataset.prefetch(tf.data.AUTOTUNE) 预取数据，以减少 I/O 瓶颈。 tf.data.AUTOTUNE 会自动调整预取缓冲区的大小。

• 并行处理 (Parallel Processing): 使用 dataset.map(..., num_parallel_calls=tf.data.AUTOTUNE) 并行地应用转换函数，以提高数据处理速度。

总结：

tf.data API 提供了多种方法来从文件读取数据，包括文本文件、CSV 文件和 TFRecord 文件。 选择合适的方法取决于你的数据格式和性能要求。 通过结合数据洗牌、批处理、预取和并行处理等高级技巧，你可以构建高效的数据输入管道，从而加速 TensorFlow 模型的训练。