使用TF Dataset训练LSTM时遇TypeError: unhashable type: 'list'错误求助

一、问题背景 #

首次尝试将TF Dataset与LSTM结合，添加多GPU优化并整理代码后，训练时触发错误：TypeError: unhashable type: 'list'，以下是问题排查和数据集处理优化建议。

二、数据处理函数 #

def csv_loader(path):
    return tf.data.experimental.CsvDataset(
        path,
        record_defaults=[tf.float32, tf.float32, tf.float32], 
        header=header,
        field_delim=',', 
        select_cols=[0,1,2,5],
        na_value='nan')

def split_feature_label(x, lbl_position, nb_attrib):
    output = x[self.input_sequence_length - 1, lbl_position + 1]
    # 移除输出列和ID列
    sub = list(range(x.shape[1]))
    sub.pop(lbl_position + 1)  
    sub.pop(0)  

    x = tf.transpose(tf.gather(tf.transpose(x), sub))
    return {'input1': x[:self.input_sequence_length, :-nb_attrib],
            'input2': x[self.input_sequence_length - 1, -nb_attrib:]}, output

def filter_mixed_csv_sample(x):
    # 过滤包含多来源ID的样本
    y, idx = tf.unique(x[:, 0])
    if len(y) > 1:
        return False
    return True

def filter_nan_missing_values(x):
    # 过滤含NaN的样本
    try:
        ynan = tf.math.is_nan(x)
        return tf.math.logical_not(tf.math.reduce_any(ynan))
    except:
        return False

def to_timeseries(x):
    # 转换为LSTM兼容的3D数据集
    x = x.map(lambda *items: tf.stack(items), num_parallel_calls=tf.data.AUTOTUNE)
    x = x.window(self.input_sequence_length + self.output_sequence_length, shift=1,
                drop_remainder=True)  
    x = x.flat_map(lambda i: i).batch(self.input_sequence_length + self.output_sequence_length)
    return x

def is_test(x, _):
    # 划分测试集
    return x % int(self.val_split * 100) == 0
        
def is_train(x, y):
    return not is_test(x, y)

def apply_scaler(x, y):
    # 特征与标签标准化
    x1_std = (x['input1'] - x1_scaler.data_min_) / (x1_scaler.data_max_ - x1_scaler.data_min_)
    x1_scaled = x1_std * (x1_scaler.feature_range[1] - x1_scaler.feature_range[0]) + x1_scaler.feature_range[0]

    x2_std = (x['input2'] - x2_scaler.data_min_) / (x2_scaler.data_max_ - x2_scaler.data_min_)
    x2_scaled = x2_std * (x1_scaler.feature_range[1] - x2_scaler.feature_range[0]) + x2_scaler.feature_range[0]

    y_std = (y - y_scaler.data_min_) / (y_scaler.data_max_ - y_scaler.data_min_)
    y_scaled = y_std * (y_scaler.feature_range[1] - y_scaler.feature_range[0]) + y_scaler.feature_range[0]
    return {'input1': x1_scaled, 'input2': x2_scaled}, y_scaled

三、数据处理流程 #

tf_list = tf.data.Dataset.list_files(list_files, shuffle=True)
dataset = tf_list.interleave(csv_loader, cycle_length=1)

with tf.device('/cpu:0'):  
    dataset = to_timeseries(self.dataset)
    dataset = dataset.ignore_errors()
    dataset = dataset.filter(filter_nan_missing_values)
    dataset = dataset.filter(filter_mixed_csv_sample)
    if shuffle:
        dataset = dataset.shuffle(shuffle_buffer)

    dataset = dataset.map(lambda x: split_feature_label(x, label_index, nb_attributs), num_parallel_calls=tf.data.AUTOTUNE)
    
    # 拆分训练/测试集
    if val_split > 0:
        recover = lambda x, y: y
        test_set = dataset.enumerate() \
            .filter(is_test) \
            .map(recover)

        trainning_set = dataset.enumerate() \
            .filter(is_train) \
            .map(recover)
    
    # 多GPU配置
    if gpu:
        strategy = tf.distribute.MirroredStrategy()
        batch_size_per_replica = batch_size * strategy.num_replicas_in_sync
    else:
        batch_size_per_replica = batch_size
    
    if val_split == 0:
        dataset = dataset.batch(batch_size_per_replica)
        dataset = dataset.cache()
        dataset = dataset.prefetch(2)  

    else:
        trainning_set = trainning_set.batch(batch_size_per_replica).cache().prefetch(2)
        test_set = test_set.batch(batch_size_per_replica).cache().prefetch(2)

# 加载标准化器并应用
x1_scaler = load('/artefacts/scalers/x1_scaler.sclr')
x2_scaler = load('/artefacts/scalers/x2_scaler.sclr')
y_scaler = load('/artefacts/scalers/y_scaler.sclr')
dataset = dataset.map(apply_scaler, num_parallel_calls=tf.data.AUTOTUNE)

四、训练代码 #

if val_split > 0:
    history = model.fit(trainning_set, validation_data=test_set,
                        verbose=verbose,
                        epochs=epochs,
                        callbacks=[checkpoint, early_stop],
                        shuffle=shuffle)
        
else:
    history = model.fit(dataset,
                        verbose=verbose,
                        epochs=epochs,
                        callbacks=[checkpoint, early_stop],
                        shuffle=shuffle)

五、错误信息 #

Exception has occurred: TypeError
unhashable type: 'list'
  File "/home/cy6112/CASTOR-S3/CASTOR-S3-cy6112/Code/timeseries_nn/core.py", line 797, in train
    history = model.fit(dataset,
              ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/cy6112/CASTOR-S3/CASTOR-S3-cy6112/Code/timeseries_nn/workbench.py", line 33, in <module>
    NN.train(ds, 200, 10, verbose=2, shuffle=True)
TypeError: unhashable type: 'list'

六、数据集结构 #

<_ParallelMapDataset element_spec=({'input1': TensorSpec(shape=(None, None, 3), dtype=tf.float32, name=None), 'input2': TensorSpec(shape=(None, 13), dtype=tf.float32, name=None)}, TensorSpec(shape=(None,), dtype=tf.float32, name=None))

错误原因排查 #

1. 多GPU策略使用不规范：创建MirroredStrategy后，未将模型构建、编译及数据集核心处理逻辑放入strategy.scope()上下文，导致分布式环境下张量结构不兼容，触发哈希错误。
2. 数据集拆分的recover函数错误：enumerate()后的数据集元素为(索引, (特征, 标签))，原recover = lambda x, y: y仅返回标签，导致训练/测试集丢失输入特征，模型无法匹配输入结构。
3. 标准化时机错误：在batch()之后执行apply_scaler，批量张量与标准化器的样本级统计量不匹配，且降低处理效率。

优化与修复方案 #

1. 修复多GPU策略使用 #

将模型构建、编译及数据集关键处理放入策略作用域：

if gpu:
    strategy = tf.distribute.MirroredStrategy()
    with strategy.scope():
        # 在此处构建并编译模型
        model = build_lstm_model()
        model.compile(optimizer='adam', loss='mse')
    batch_size_per_replica = batch_size * strategy.num_replicas_in_sync
else:
    batch_size_per_replica = batch_size
    model = build_lstm_model()
    model.compile(optimizer='adam', loss='mse')

2. 修正数据集拆分逻辑 #

调整recover函数保留完整的(特征, 标签)结构：

if val_split > 0:
    recover = lambda idx, data: data  # 保留特征与标签对
    test_set = dataset.enumerate() \
        .filter(is_test) \
        .map(recover, num_parallel_calls=tf.data.AUTOTUNE)

    trainning_set = dataset.enumerate() \
        .filter(is_train) \
        .map(recover, num_parallel_calls=tf.data.AUTOTUNE)

3. 调整标准化时机 #

将apply_scaler移至split_feature_label之后、batch()之前：

dataset = dataset.map(lambda x: split_feature_label(x, label_index, nb_attributs), num_parallel_calls=tf.data.AUTOTUNE)
# 先标准化，再批量处理
dataset = dataset.map(apply_scaler, num_parallel_calls=tf.data.AUTOTUNE)
# 后续再执行拆分、batch等操作

4. 性能优化 #

• 简化过滤函数：移除filter_nan_missing_values中的try-except，TensorFlow原生函数可稳定处理张量：

  def filter_nan_missing_values(x):
      ynan = tf.math.is_nan(x)
      return tf.math.logical_not(tf.math.reduce_any(ynan))
  

• 优化ID过滤逻辑：用reduce_all替代tf.unique，减少张量创建开销：

  def filter_mixed_csv_sample(x):
      first_id = x[0, 0]
      return tf.reduce_all(tf.equal(x[:, 0], first_id))
  

• 高效生成时间序列：直接在window后批量处理，避免多余的flat_map：

  def to_timeseries(x):
      x = x.map(lambda *items: tf.stack(items), num_parallel_calls=tf.data.AUTOTUNE)
      window_size = self.input_sequence_length + self.output_sequence_length
      x = x.window(window_size, shift=1, drop_remainder=True)
      x = x.flat_map(lambda window: window.batch(window_size))
      return x
  

• 缓存策略调整：将cache()放在所有过滤、映射操作之后，缓存处理后的样本，避免重复计算：

  dataset = dataset.filter(filter_nan_missing_values)
  dataset = dataset.filter(filter_mixed_csv_sample)
  dataset = dataset.map(...)
  dataset = dataset.cache()
  if shuffle:
      dataset = dataset.shuffle(shuffle_buffer)
  dataset = dataset.batch(...)
  dataset = dataset.prefetch(tf.data.AUTOTUNE)
  

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