Scikit-learn Pipeline处理加州住房数据集时测试集转换报错问题排查

AI 小北

2026-4-27

问题排查与解决方案

从你的代码来看，训练集拟合正常但测试集转换报错，最可能的原因是**get_columns()函数中定义的列列表包含了后续AddComboFeatures生成的新特征，而Scikit-learn的ColumnTransformer在某些场景下（尤其是旧版本）会在初始化阶段尝试验证这些列是否存在于原始输入数据中，或者在测试集转换时由于列索引匹配问题导致找不到这些新特征**。

核心问题分析

你的log_like_columns和numerical_columns包含了population_per_room、income_per_house这类由AddComboFeatures动态生成的列，这些列在原始输入数据（训练集/测试集的初始数据）中是不存在的。虽然训练集拟合时，add_features步骤会先生成这些列，再传递给ColumnTransformer，所以不会报错；但测试集转换时，ColumnTransformer在拟合阶段记录的列索引是基于训练集处理后的列顺序，若测试集处理后的列顺序出现偏差（比如极端情况下的列名排序变化），就会导致索引不匹配，进而抛出找不到列的错误。

修复步骤

我们可以通过解耦原始列与新生成列的定义，并明确告诉ColumnTransformer哪些列需要处理，来解决这个问题：

1. 修正列定义函数

修改get_columns()，将原始列和新生成的列分开定义，再合并，确保所有要处理的列都是add_features步骤后存在的列：

def get_columns():
    # 原始数据中的列
    log_like_original = [
        "total_rooms", "total_bedrooms", "population", "households", "median_income",
    ]
    numerical_original = [
        "longitude", "latitude", "housing_median_age", "total_rooms", "total_bedrooms",
        "population", "households", "median_income",
    ]
    remaining_original = np.array(numerical_original)[
        np.isin(numerical_original, log_like_original, invert=True)
    ]
    categorical_columns = ["ocean_proximity"]
    
    # AddComboFeatures生成的新列
    log_like_new = [
        "population_per_room", "income_per_house", "income_per_population",
    ]
    remaining_new = ["rooms_per_bedroom"]
    
    # 合并得到最终要处理的列列表
    log_like_columns = log_like_original + log_like_new
    remaining_columns = list(remaining_original) + remaining_new
    
    return log_like_columns, remaining_columns, categorical_columns

2. 优化特征生成类的列检查逻辑

修改AddComboFeatures的transform方法，用更清晰的方式检查必要列是否存在，避免numpy操作可能带来的隐式错误：

class AddComboFeatures(BaseEstimator, TransformerMixin):
    def fit(self, X, y=None):
        return self
    def transform(self, X, y=None):
        X_copy = X.copy()
        required_columns = [
            "total_rooms", "total_bedrooms", "population", "median_income", "households",
        ]
        # 检查必要列是否存在
        missing_columns = [col for col in required_columns if col not in X_copy.columns]
        if missing_columns:
            raise ValueError(f"The following columns are missing: {', '.join(missing_columns)}")
        # 计算组合特征
        X_copy["rooms_per_bedroom"] = safe_divide(X_copy["total_rooms"], X_copy["total_bedrooms"])
        X_copy["population_per_room"] = safe_divide(X_copy["population"], X_copy["total_rooms"])
        X_copy["income_per_house"] = safe_divide(X_copy["median_income"], X_copy["households"])
        X_copy["income_per_population"] = safe_divide(X_copy["median_income"], X_copy["population"])
        return X_copy
    def get_feature_names_out(self, input_features=None):
        return list(input_features) + [
            "rooms_per_bedroom", "population_per_room", "income_per_house", "income_per_population",
        ]

3. 验证测试集转换流程

在测试代码中，先单独验证add_features步骤是否正确生成了所有新列，再执行完整的Pipeline转换：

import data
import features
from sklearn.model_selection import train_test_split

housing_path = r"Path\projects\housing_project_ex\data\housing.csv"
df_housing = data.load_data(housing_path)
df_housing = data.clean_data(df_housing)

# 正确拆分特征和标签
df_housing_X = df_housing.drop(["median_house_value"], axis=1)
df_housing_y = df_housing["median_house_value"]
train_X_raw, test_X_raw, train_labels, test_labels = train_test_split(df_housing_X, df_housing_y, random_state=42)

# 先验证add_features步骤
pipeline = features.get_pipeline()
add_features_step = pipeline.named_steps["add_features"]
test_X_added = add_features_step.transform(test_X_raw)
print("Generated columns after add_features:", test_X_added.columns.tolist())

# 再执行完整转换
train_X = pipeline.fit_transform(train_X_raw)
test_X = pipeline.transform(test_X_raw)
print("Test set transformation successful!")
print(test_X.info())

额外注意事项

确保你的clean_data函数已经处理了所有缺失值（比如填充total_bedrooms的NaN），避免safe_divide中出现除以NaN的情况。
如果使用Scikit-learn 1.2+版本，推荐使用set_output(transform="pandas")来保持输出为DataFrame，方便调试列名问题。

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