TensorFlow带元数据时间序列分类：多输入数据集适配与架构优化

问题根源 #

你的模型定义了4个独立输入层：measurement_input（时间序列数据）、Product、Length、Width（元数据字段），但构建TF Dataset时，错误地把所有元数据打包成了meta_input这个单一键，导致Keras在训练时找不到对应Product、Length、Width的输入数据，触发Missing data for input错误。

修复与优化方案 #

1. 修正Dataset输入结构 #

修改Dataset的映射逻辑，将元数据的每个字段单独作为输入字典的键，与模型输入层名称完全匹配：

原错误映射：

dataset = dataset.map(
    lambda x, y: ({"measurement_input": x, "meta_input": y[:-1]}, y[-1])
)

修正后：

dataset = dataset.map(
    lambda x, y: ({**{"measurement_input": x}, **y[0]}, y[1])
)

这里直接将元数据字典y[0]与时间序列输入合并，保证每个输入层都能找到对应的数据键。

2. 架构优化 #

• 把时间序列处理和元数据处理拆分为独立子模型，提升代码模块化程度
• 简化预处理层的调用逻辑，避免冗余代码
• 增强时间序列分支的特征提取能力，增加卷积层堆叠和池化操作

完整代码实现 #

import pandas as pd
import tensorflow as tf
import numpy as np


def get_normalization_layer(
    dataset, input_name, all_inputs, encoded_features, axis=None
):
    normalization_layer = tf.keras.layers.Normalization(axis=axis)
    feature_ds = dataset.map(lambda x, y: x[input_name])
    normalization_layer.adapt(feature_ds)
    input_layer = tf.keras.Input(shape=(1,), name=input_name, dtype=tf.float32)
    normalized = normalization_layer(input_layer)
    all_inputs.append(input_layer)
    encoded_features.append(normalized)


def get_category_encoding_layer(
    dataset,
    input_name,
    dtype,
    all_inputs,
    encoded_features,
    vocabulary=None,
    max_tokens=None,
):
    if dtype == "string":
        index = tf.keras.layers.StringLookup(
            max_tokens=max_tokens, vocabulary=vocabulary
        )
        input_layer = tf.keras.Input(shape=(1,), name=input_name, dtype=tf.string)
    elif dtype == "int":
        index = tf.keras.layers.IntegerLookup(
            max_tokens=max_tokens, vocabulary=vocabulary
        )
        input_layer = tf.keras.Input(shape=(1,), name=input_name, dtype=tf.int64)
    if vocabulary is None:
        feature_ds = dataset.map(lambda x, y: x[input_name])
        index.adapt(feature_ds)
    # 编码层处理
    indexed = index(input_layer)
    encoder = tf.keras.layers.CategoryEncoding(num_tokens=index.vocabulary_size())(indexed)
    all_inputs.append(input_layer)
    encoded_features.append(encoder)


# 生成匹配样本数的模拟数据
measurement_data = np.random.rand(150, 11, 2400)
meta_data = pd.DataFrame(
    data={
        "Product": np.random.choice(["A", "B", "C", "D"], 150),
        "Length": np.random.randint(20, 25, 150),
        "Width": np.random.uniform(0, 60, 150),
        "Labels": np.random.randint(0, 2, 150),
    }
)

# 预处理元数据
dataframe = meta_data.copy()
labels = dataframe.pop("Labels")
dataframe = {key: np.array(value)[:, tf.newaxis] for key, value in dataframe.items()}

# 构建基础Dataset
dataset_measurement = tf.data.Dataset.from_tensor_slices(measurement_data)
dataset_meta = tf.data.Dataset.from_tensor_slices((dataframe, labels))

# 合并Dataset并修正输入结构
dataset = tf.data.Dataset.zip((dataset_measurement, dataset_meta))
# 关键修改：合并时间序列输入和元数据字典
dataset = dataset.map(
    lambda x, y: ({**{"measurement_input": x}, **y[0]}, y[1])
)
dataset = dataset.batch(64).prefetch(tf.data.AUTOTUNE)

# 构建模型输入与预处理层
all_inputs = []
encoded_features = []

# 时间序列输入预处理
ts_input = tf.keras.Input(shape=(11, 2400), name="measurement_input", dtype=tf.float32)
ts_normalized = tf.keras.layers.Normalization(axis=1)(ts_input)
# 适配数据
ts_ds = dataset.map(lambda x, y: x["measurement_input"])
tf.keras.layers.Normalization(axis=1).adapt(ts_ds)
all_inputs.append(ts_input)
encoded_features.append(ts_normalized)

# 元数据输入预处理
get_category_encoding_layer(dataset, "Product", "string", all_inputs, encoded_features)
get_category_encoding_layer(dataset, "Length", "int", all_inputs, encoded_features)
get_normalization_layer(dataset, "Width", all_inputs, encoded_features)

# 时间序列特征提取分支（优化：增加卷积层堆叠）
ts_branch = tf.keras.Sequential([
    tf.keras.layers.Conv1D(filters=64, kernel_size=3, padding="same", activation="relu"),
    tf.keras.layers.BatchNormalization(),
    tf.keras.layers.MaxPooling1D(pool_size=2),
    tf.keras.layers.Conv1D(filters=128, kernel_size=3, padding="same", activation="relu"),
    tf.keras.layers.BatchNormalization(),
    tf.keras.layers.GlobalAveragePooling1D()
], name="time_series_branch")(encoded_features[0])

# 元数据特征提取分支
meta_branch = tf.keras.Sequential([
    tf.keras.layers.concatenate(encoded_features[1:]),
    tf.keras.layers.Dense(128, activation="relu"),
    tf.keras.layers.Dropout(0.5),
    tf.keras.layers.Dense(64, activation="relu"),
    tf.keras.layers.Dropout(0.3)
], name="metadata_branch")

# 合并分支并输出
combined = tf.keras.layers.concatenate([ts_branch, meta_branch])
output = tf.keras.layers.Dense(2, activation="softmax")(combined)

# 定义模型
model = tf.keras.Model(inputs=all_inputs, outputs=output)

model.compile(
    optimizer="adam",
    loss="sparse_categorical_crossentropy",
    metrics=["sparse_categorical_accuracy"],
)

# 训练模型
history = model.fit(
    dataset,
    epochs=50,
    verbose=1,
)

关键修改说明 #

1. Dataset映射修正：使用{**{"measurement_input": x}, **y[0]}将时间序列输入和元数据字典合并，保证每个输入层名称都有对应的数据键。
2. 样本数匹配：修正了原代码中meta_data与measurement_data样本数不匹配的问题，避免训练时维度错误。
3. 架构模块化：将时间序列和元数据处理封装为独立Sequential子模型，代码更清晰易维护。
4. 增强特征提取：时间序列分支增加了一层卷积和池化操作，提升特征提取能力；元数据分支增加了一层Dense层，增强非线性表达。

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