如何使用sklearn.Pipeline实现仅训练集步骤与类df.apply功能

1. 实现仅对训练数据生效的处理步骤 #

你可以通过继承BaseEstimator和TransformerMixin自定义转换器，在fit方法中实现训练阶段专属逻辑，transform方法直接返回原数据即可实现仅训练生效的效果。
如果你的训练专属操作需要修改样本标签/数量（如过采样、欠采样），建议搭配imblearn.pipeline.Pipeline使用，支持中间步骤接收并修改y值。
代码示例（以训练阶段欠采样为例）：

from sklearn.base import BaseEstimator, TransformerMixin
from imblearn.under_sampling import RandomUnderSampler

class TrainOnlyUnderSampler(BaseEstimator, TransformerMixin):
    def fit(self, X, y=None):
        # 仅在fit阶段执行训练集专属处理逻辑
        self.sampler = RandomUnderSampler(random_state=42)
        self.sampler.fit_resample(X, y)
        return self
    
    def transform(self, X, y=None):
        # 测试/推理阶段不做任何处理，直接返回原数据
        return X

2. 实现类似pandas df.apply的自定义处理功能 #

直接使用sklearn.preprocessing.FunctionTransformer封装你的自定义apply逻辑即可，完全兼容Pipeline的调用规范。
代码示例：

from sklearn.preprocessing import FunctionTransformer
import numpy as np
import pandas as pd

# 自定义apply逻辑函数，输入输出均为DataFrame/数组
def custom_apply_logic(X):
    # 此处可写入任意你原本在df.apply中实现的处理逻辑
    return X.apply(np.log1p, axis=1)

# 封装为Pipeline可用的转换器
apply_transformer = FunctionTransformer(custom_apply_logic)

完整流水线示例 #

将上述两个自定义转换器加入Pipeline即可：

from imblearn.pipeline import Pipeline
from sklearn.ensemble import RandomForestClassifier

clf_pipeline = Pipeline([
    ("custom_apply", apply_transformer), # 自定义apply逻辑，训练、测试阶段均生效
    ("train_only_sampler", TrainOnlyUnderSampler()), # 仅训练阶段生效的采样步骤
    ("classifier", RandomForestClassifier(random_state=42))
])

# 训练时会执行所有步骤的fit逻辑，预测时仅执行所有步骤的transform逻辑，自动区分训练测试处理规则
clf_pipeline.fit(X_train, y_train)
y_pred = clf_pipeline.predict(X_test)

注意事项 #

• 如果你的自定义处理需要用到训练集的统计量（如自定义分箱阈值、标准化均值），不要直接用FunctionTransformer，需继承BaseEstimator在fit方法中存储统计量，transform阶段调用存储的统计量处理数据，避免数据泄露。
• 普通sklearn原生Pipeline不支持中间步骤修改y值，涉及标签、样本数量修改的训练专属操作请使用imblearn的Pipeline。

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