如何在同一ColumnTransformer流水线中使用其生成的新列？

ColumnTransformer的核心逻辑是每个子转换器分支独立读取原始数据集的指定列，而非前一个分支处理后的结果。你定义的travel_category_pipeline拿到的是原始数据里的列，而total_relatives是total_relatives_pipeline生成的新列——原始数据里根本没有这个列，自然会报错。同理，traveling_category和age_interval这类其他分支生成的新列，后续分支也没法直接读取，因为它们没有回写到原始数据集供其他分支调用。

方案1：把依赖步骤合并到同一个Pipeline里 #

把需要生成新列、并基于该列做后续处理的逻辑打包到同一个Pipeline分支中，这样就能在同一个流程里复用新生成的列。

修改后的示例代码：

import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import OrdinalEncoder, FunctionTransformer, SimpleImputer, OneHotEncoder, StandardScaler

# 合并亲属数计算与旅行类别划分逻辑
def sum_and_categorize_relatives(X):
    X_copy = X.copy()
    # 第一步：生成total_relatives
    X_copy['total_relatives'] = X_copy['SibSp'] + X_copy['Parch']
    # 第二步：基于total_relatives生成traveling_category
    conditions = [
        (X_copy['total_relatives'] == 0),
        (X_copy['total_relatives'] >= 1) & (X_copy['total_relatives'] <= 3),
        (X_copy['total_relatives'] >= 4)
    ]
    categories = ['A', 'B', 'C']
    X_copy['traveling_category'] = np.select(conditions, categories, default='Unknown')
    # 返回需要后续处理的列
    return X_copy[['traveling_category']]

def travel_cat_name(function_transformer, feature_names_in):
    return ["traveling_category"]

# 整合年龄处理，返回age_interval列
def age_transformer(X):
    X_copy = X.copy()
    median_age_by_class = X_copy.groupby('Pclass')['Age'].median().reset_index()
    median_age_by_class.columns = ['Pclass', 'median_age']
    for index, row in median_age_by_class.iterrows():
        class_value = row['Pclass']
        median_age = row['median_age']
        X_copy.loc[X_copy['Pclass'] == class_value, 'Age'] = X_copy.loc[X_copy['Pclass'] == class_value, 'Age'].fillna(median_age)
    bins = [0, 10, 20, 30, 40, 50, 60, 70, 100]
    X_copy['age_interval'] = pd.cut(X_copy['Age'], bins=bins)
    return X_copy[['age_interval']]

def interval_name(function_transformer, feature_names_in):
    return ["age_interval"]

class_order = [[1, 2, 3]]
ord_pipeline = make_pipeline(OrdinalEncoder(categories=class_order))

# 合并后的亲属处理流水线
relatives_pipeline = make_pipeline(
    FunctionTransformer(sum_and_categorize_relatives, feature_names_out=travel_cat_name)
)

age_pipeline = make_pipeline(
    FunctionTransformer(age_transformer, feature_names_out=interval_name)
)

cat_pipeline = make_pipeline(
    SimpleImputer(strategy="most_frequent"),
    OneHotEncoder(handle_unknown="ignore")
)

num_pipeline = make_pipeline(StandardScaler())

preprocessing = ColumnTransformer([
    ("ord", ord_pipeline, ['Pclass']),
    ("age_processing", age_pipeline, ['Pclass', 'Age']),
    ("relatives", relatives_pipeline, ['SibSp', 'Parch']),
    ("cat", cat_pipeline, ['Sex', 'Embarked']),  # 只传原始分类列，新生成的分类列已由对应分支处理
    ("num", num_pipeline, ['Fare']),
])

方案2：先统一生成所有新列，再做分类型处理 #

先通过一个Pipeline生成所有需要的新列，把原始列和新列合并成完整数据集，再用ColumnTransformer对最终数据集做分类型处理。

示例代码：

import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import make_pipeline, Pipeline
from sklearn.preprocessing import OrdinalEncoder, FunctionTransformer, SimpleImputer, OneHotEncoder, StandardScaler

def generate_all_new_columns(X):
    X_copy = X.copy()
    # 生成total_relatives
    X_copy['total_relatives'] = X_copy['SibSp'] + X_copy['Parch']
    # 生成traveling_category
    conditions = [
        (X_copy['total_relatives'] == 0),
        (X_copy['total_relatives'] >= 1) & (X_copy['total_relatives'] <= 3),
        (X_copy['total_relatives'] >= 4)
    ]
    categories = ['A', 'B', 'C']
    X_copy['traveling_category'] = np.select(conditions, categories, default='Unknown')
    # 处理年龄并生成age_interval
    median_age_by_class = X_copy.groupby('Pclass')['Age'].median().reset_index()
    median_age_by_class.columns = ['Pclass', 'median_age']
    for index, row in median_age_by_class.iterrows():
        class_value = row['Pclass']
        median_age = row['median_age']
        X_copy.loc[X_copy['Pclass'] == class_value, 'Age'] = X_copy.loc[X_copy['Pclass'] == class_value, 'Age'].fillna(median_age)
    bins = [0, 10, 20, 30, 40, 50, 60, 70, 100]
    X_copy['age_interval'] = pd.cut(X_copy['Age'], bins=bins)
    return X_copy

# 第一步：生成所有新列，扩充原始数据集
data_enrich_pipeline = make_pipeline(
    FunctionTransformer(generate_all_new_columns)
)

class_order = [[1, 2, 3]]
ord_pipeline = make_pipeline(OrdinalEncoder(categories=class_order))

cat_pipeline = make_pipeline(
    SimpleImputer(strategy="most_frequent"),
    OneHotEncoder(handle_unknown="ignore")
)

num_pipeline = make_pipeline(StandardScaler())

# 第二步：用ColumnTransformer处理包含所有新列的完整数据集
preprocessing = Pipeline([
    ("enrich", data_enrich_pipeline),
    ("transform", ColumnTransformer([
        ("ord", ord_pipeline, ['Pclass']),
        ("cat", cat_pipeline, ['Sex', 'Embarked', 'traveling_category', 'age_interval']),
        ("num", num_pipeline, ['Fare', 'total_relatives']),
    ]))
])

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