TensorFlow中TFRecords时间序列窗口化数据集管道报错排查

问题背景 #

编写TensorFlow时序预测模型时，基于TFRecords文件（每个文件对应一次试验，含图像与标签）构建窗口化数据集管道，运行时报错：

ValueError: Input serialized must be a scalar

用户提供的核心代码如下：

单样本解析函数 #

IMG_SIZE_INPUT = (60, 80, 1)

WINDOW_SIZE = 10
BATCH_SIZE = 32

def parse_single_example(example):
    feature_description = {
        'timestamp': tf.io.FixedLenFeature([], tf.int64),
        'image_raw': tf.io.FixedLenFeature([], tf.string),
        'label': tf.io.FixedLenFeature([], tf.int64)
    }

    features = tf.io.parse_single_example(example, feature_description)

    b_image = features['image_raw'] # get byte string
    image = tf.io.parse_tensor(b_image, out_type = tf.uint8)
    image = tf.reshape(image, IMG_SIZE_INPUT)
    image = tf.cast(image, tf.float32)

    timestamp = features['timestamp']
    label = features['label']

    return timestamp, image, label

数据集管道函数 #

def _parse_and_augment_image(example, seed, do_augment):
    timestamp, image, label = parse_single_example(example)

    if do_augment:
        # Pad the image and the mask to apply later a crop
        image = tf.image.resize_with_crop_or_pad(image, IMG_SIZE_INPUT[0] + 24, IMG_SIZE_INPUT[1] + 24)

        # Make a new seed.
        new_seed = tf.random.experimental.stateless_split(seed, num = 1)[0, :]

        # Random crop back to the original size.
        image = tf.image.stateless_random_crop(
            image, size = IMG_SIZE_INPUT, seed = new_seed)

        # Random flip L/R
        image = tf.image.stateless_random_flip_left_right(image, new_seed)

        # Random flip U/D
        image = tf.image.stateless_random_flip_up_down(image, new_seed)

        # Random brightness.
        image = tf.image.stateless_random_brightness(
            image, max_delta = 0.075 * 255, seed = new_seed)

        image = tf.clip_by_value(image, 0, 255)

    return timestamp, image, label

def prepare_for_training(tf_record_single_file_path, batch_size, window_size = WINDOW_SIZE, shift_size = 1):
    dataset = tf.data.TFRecordDataset(tf_record_single_file_path)

    # Windowing
    windowed_dataset = dataset.window(window_size, shift=shift_size, drop_remainder=True)
    dataset = windowed_dataset.flat_map(lambda window: window.batch(window_size))

    # Parsing
    counter = tf.data.experimental.Counter()
    train_dataset = tf.data.Dataset.zip((dataset, (counter, counter)))
    dataset = train_dataset.map(partial(_parse_and_augment_image, do_augment = False), num_parallel_calls=10)

    # Batching
    dataset = dataset.batch(batch_size)

    return dataset


def read_dataset(filename_dir, batch_size):
    all_tf_records = glob.glob(filename_dir + os.sep + "*.tfrecords")

    # Test on one file
    dataset = prepare_for_training(all_tf_records[0], 2)

错误原因 #

prepare_for_training函数操作顺序错误：先执行窗口化与window.batch(window_size)，此时数据集的每个元素是包含window_size个序列化TFRecord样本的批量张量，但parse_single_example仅能处理单个标量的序列化样本，传入批量数据直接触发类型不匹配错误。

修正后的解决方案 #

调整操作顺序：先解析并增强单个样本，再对解析后的样本做窗口化，最后打包训练批次。修正后的prepare_for_training函数如下：

from functools import partial
import tensorflow as tf
import glob
import os

IMG_SIZE_INPUT = (60, 80, 1)
WINDOW_SIZE = 10
BATCH_SIZE = 32

# 保持原parse_single_example和_parse_and_augment_image函数不变

def prepare_for_training(tf_record_single_file_path, batch_size, window_size = WINDOW_SIZE, shift_size = 1):
    dataset = tf.data.TFRecordDataset(tf_record_single_file_path)

    # 1. 先解析单个样本并处理增强
    counter = tf.data.experimental.Counter()
    train_dataset = tf.data.Dataset.zip((dataset, (counter, counter)))
    parsed_dataset = train_dataset.map(partial(_parse_and_augment_image, do_augment=False), num_parallel_calls=10)

    # 2. 对解析后的单个样本做窗口化
    windowed_dataset = parsed_dataset.window(window_size, shift=shift_size, drop_remainder=True)
    # 将每个窗口转换为包含window_size个样本的批次
    window_batch_dataset = windowed_dataset.flat_map(lambda window: window.batch(window_size))

    # 3. 补充：将窗口数据拆分为输入序列与预测目标（适配时序预测需求）
    def split_window(window):
        timestamps, images, labels = window
        # 输入为前window_size-1个样本，目标为最后一个标签
        return (timestamps[:-1], images[:-1]), labels[-1]
    
    window_batch_dataset = window_batch_dataset.map(split_window)

    # 4. 打包为训练批次并优化加载效率
    dataset = window_batch_dataset.batch(batch_size).prefetch(tf.data.AUTOTUNE)

    return dataset

关键说明 #

1. 操作顺序调整：确保parse_single_example始终处理单个标量序列化样本，避免批量输入
2. 时序适配：新增split_window函数，将每个窗口拆分为输入序列与预测目标，贴合时序预测场景需求
3. 性能优化：添加prefetch(tf.data.AUTOTUNE)提升数据加载效率

内容的提问来源于stack exchange，提问作者PMDP3