没有 Pipeline 的机器学习代码通常在 Jupyter Notebook 中散落为以下碎片：StandardScaler().fit_transform(X_train) 在一行，model.fit() 在翻了好几页后的另一个 Cell 里，预处理和训练之间被分析和可视化的代码隔开。更致命的是，交叉验证时很容易把 fit_transform 应用于整个数据集而非仅训练集，导致测试集信息泄漏到训练过程中——这种 Bug 不会报错，只会让交叉验证分数虚高，而模型上线后的真实表现远低于预期。scikit-learn 的 Pipeline 设计正是为了解决这类工程化问题。

---

一、scikit-learn 的统一接口设计

scikit-learn 的 API 稳定性在 Python 生态中相当突出——它的三大接口 fit、transform、predict 已经稳定运行了 15 年以上。这套接口将所有的数据操作和模型训练统一为三类角色：

角色	接口	代表类
Estimator	fit(X, y)	所有模型、Scaler、Encoder
Transformer	fit(X, y).transform(X) 或 fit_transform(X)	StandardScaler、OneHotEncoder、PCA
Predictor	predict(X) / predict_proba(X)	LogisticRegression、RandomForest、SVC

Pipeline 的核心价值在于将多个 Transformer 和一个最终的 Predictor 串联为一个整体对象——对外只暴露 fit 和 predict，内部自动按顺序调用各个步骤的 fit_transform。

---

二、Pipeline 基础：防止数据泄漏的最小单元

from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score

pipe = make_pipeline(StandardScaler(), LogisticRegression())
scores = cross_val_score(pipe, X_train, y_train, cv=5)

# 等价于以下手动流程，但完全避免数据泄漏：
# 对每一折：scaler.fit_transform(train_fold) → model.fit() → scaler.transform(test_fold) → model.predict()

错误做法中，StandardScaler.fit_transform(X_all) 使用全量数据（包括测试集）计算均值和标准差，这个信息随后被带入到交叉验证的每一折中——测试集的数据特征"污染"了训练过程。Pipeline 的正确做法是将 Scaler.fit_transform 限定在每折的训练集内，对测试集仅执行 transform。

---

三、ColumnTransformer：混合类型特征的一站式处理

真实数据集的列通常混合了数值、类别和文本类型。ColumnTransformer 允许为不同类型的列配置独立的处理管线：

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression

numeric_features = ["age", "fare", "sibsp", "parch"]
categorical_features = ["sex", "embarked", "pclass"]
text_features = ["name"]

preprocessor = ColumnTransformer([
    ("num", StandardScaler(), numeric_features),
    ("cat", OneHotEncoder(handle_unknown="ignore"), categorical_features),
    ("txt", TfidfVectorizer(max_features=100), text_features),
])

pipe = Pipeline([
    ("preprocessor", preprocessor),
    ("classifier", LogisticRegression(max_iter=1000)),
])

pipe.fit(X_train, y_train)

ColumnTransformer 内部并行处理三组特征：数值列走 StandardScaler，类别列走 OneHotEncoder，文本列走 TfidfVectorizer。处理后的结果自动水平拼接，送入后续的分类器。整个过程被封装在 Pipeline 中，对外只有 fit 和 predict 两个入口。

FeatureUnion：并行特征管线

from sklearn.pipeline import FeatureUnion
from sklearn.decomposition import PCA

combined_features = FeatureUnion([
    ("original", preprocessor),
    ("pca", Pipeline([
        ("select_num", ColumnTransformer([("num", StandardScaler(), numeric_features)])),
        ("pca", PCA(n_components=5)),
    ])),
])

FeatureUnion 是 ColumnTransformer 的"并行版"——它允许多条独立的处理管线并行执行，最终将各条管线的输出水平拼接。常见应用：原始特征 + 降维特征（PCA）+ 多项式交互特征三条管线并行，然后送入模型。

---

四、自定义 Transformer

当内置的 Transformer 无法满足需求时，scikit-learn 提供了两种轻量级的扩展方式：

FunctionTransformer：一行代码的自定义

from sklearn.preprocessing import FunctionTransformer
import numpy as np

log_transformer = FunctionTransformer(np.log1p, inverse_func=np.expm1)
# fit_transform(X) 直接调用 np.log1p
# inverse_transform(X) 调用 np.expm1

继承 BaseEstimator + TransformerMixin：有状态的自定义转换器

from sklearn.base import BaseEstimator, TransformerMixin
import pandas as pd

class DateFeatureExtractor(BaseEstimator, TransformerMixin):
    """从日期列提取年/月/日/星期/是否周末"""
    def __init__(self, date_column: str, drop_original: bool = True):
        self.date_column = date_column
        self.drop_original = drop_original
    
    def fit(self, X, y=None):
        return self  # 无状态，无需学习参数
    
    def transform(self, X):
        X = X.copy()
        dt = pd.to_datetime(X[self.date_column])
        X["year"]           = dt.dt.year
        X["month"]          = dt.dt.month
        X["day"]            = dt.dt.day
        X["dayofweek"]      = dt.dt.dayofweek
        X["is_weekend"]     = dt.dt.dayofweek.isin([5, 6]).astype(int)
        X["quarter"]        = dt.dt.quarter
        
        # 周期性编码：小时/月份等循环特征的标准处理方式
        X["month_sin"]      = np.sin(2 * np.pi * dt.dt.month / 12)
        X["month_cos"]      = np.cos(2 * np.pi * dt.dt.month / 12)
        
        if self.drop_original:
            X = X.drop(columns=[self.date_column])
        return X

class OutlierClipper(BaseEstimator, TransformerMixin):
    """基于 IQR 的异常值裁剪"""
    def __init__(self, factor: float = 1.5):
        self.factor = factor
        self.bounds_ = {}
    
    def fit(self, X, y=None):
        for col in X.select_dtypes("number").columns:
            q1 = X[col].quantile(0.25)
            q3 = X[col].quantile(0.75)
            iqr = q3 - q1
            self.bounds_[col] = (
                q1 - self.factor * iqr,
                q3 + self.factor * iqr,
            )
        return self
    
    def transform(self, X):
        X = X.copy()
        for col, (lower, upper) in self.bounds_.items():
            X[col] = X[col].clip(lower, upper)
        return X

自定义 Transformer 的关键约定：

• fit() 返回 self——这是 scikit-learn 接口的核心契约，保障 Pipeline 内步骤的连贯性。
• transform() 接收并返回与输入形状兼容的数据（DataFrame 或 ndarray）。
• 参数通过 __init__ 传入并作为实例属性存储，这使得后续 GridSearchCV 可以直接搜索 Transformer 的超参。

---

五、Pipeline 中的超参搜索

Pipeline 中的参数命名遵循 步骤名__参数名 的规则：

from sklearn.model_selection import GridSearchCV

pipe = Pipeline([
    ("preprocessor", ColumnTransformer([
        ("num", StandardScaler(), numeric_features),
        ("cat", OneHotEncoder(), categorical_features),
    ])),
    ("classifier", LogisticRegression()),
])

param_grid = {
    "classifier__C": [0.01, 0.1, 1, 10],
    "classifier__penalty": ["l1", "l2"],
    "preprocessor__num__with_mean": [True, False],
}

grid = GridSearchCV(pipe, param_grid, cv=5, scoring="accuracy")
grid.fit(X_train, y_train)

preprocessor__num__with_mean 这个参数名的含义是：Pipeline 中名为 preprocessor 的步骤 → 其中名为 num 的子转换器 → 其 with_mean 参数。这套命名规则将任意深度的嵌套结构映射为扁平的双下划线命名空间，使得 GridSearchCV 可以搜索 Pipeline 内任意位置的参数。

---

六、实战：泰坦尼克号生存预测的完整 Pipeline

以下 Pipeline 整合了缺失值填充、类别编码、数值标准化、自定义日期特征提取和分类器，是整个工作流的"单一事实来源"：

from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer, make_column_selector
from sklearn.preprocessing import StandardScaler, OneHotEncoder, FunctionTransformer
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score

numeric_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("clipper", OutlierClipper(factor=1.5)),
    ("scaler", StandardScaler()),
])

categorical_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
])

preprocessor = ColumnTransformer([
    ("num", numeric_transformer, make_column_selector(dtype_include="number")),
    ("cat", categorical_transformer, make_column_selector(dtype_include="object")),
    ("date", DateFeatureExtractor("ticket_date"), ["ticket_date"]),
])

full_pipeline = Pipeline([
    ("preprocessor", preprocessor),
    ("classifier", LogisticRegression(max_iter=1000, C=0.1)),
])

scores = cross_val_score(full_pipeline, X, y, cv=5, scoring="accuracy")
print(f"CV Accuracy: {scores.mean():.4f} ± {scores.std():.4f}")

这个 Pipeline 的关键收益：

• make_column_selector 自动按数据类型选择列，避免手动列出所有的数值列和类别列。
• 每列类别列内部也有 Pipeline（填充 → 编码），ColumnTransformer 管理列组分配，外层 Pipeline 串联预处理和模型。
• 交叉验证时，整个 Pipeline 作为一个黑盒被传入 cross_val_score，每折内部的填充器、缩放器、编码器都独立拟合——数据泄漏风险为零。
• 从训练到预测只需两行：pipe.fit(X_train, y_train) → pipe.predict(X_test)。

---

小结

Pipeline 不是 sklearn 的附加功能——它是构建可复现、可维护的机器学习工作流的基础设施。它的核心收益体现在三个层面：防止交叉验证中的数据泄漏（每折独立 fit_transform）；将分散在 Notebook 各处的预处理步骤收敛为单一对象（可序列化、可部署）；利用双下划线命名规则无缝接入超参搜索（GridSearchCV/Optuna）。

搭配 ColumnTransformer 处理混合类型特征、FeatureUnion 实现并行特征提取、自定义 Transformer 封装业务逻辑——这四件套构成了机器学习工程化的标准范式。泰坦尼克号实战演示的不仅是模型本身，而是"如何把一个完整的数据处理+训练链路封装为可复现的 Pipeline 对象"。

此前专栏关于 Pandas 工程化、特征可视化和数据管道编排的文章，为本文提供了从数据预处理到特征工程的上游支撑。如果本文对 ML 工作流的工程化实践有所启发，欢迎点赞、收藏与关注。