Scikit-learn 机器学习 Pipeline 构建

FreeGuideOnline 最新 2026-07-08

Scikit-learn 机器学习 Pipeline 构建完全指南

引言

在机器学习项目中，从数据预处理、特征工程到模型训练，每一步都极易出错且难以复现。Scikit-learn 的 Pipeline 工具能够将这些步骤串联成一个整体对象，极大简化代码、防止数据泄漏，并让超参数调优、模型部署变得更加可靠。本教程将带你从零开始，系统掌握 Pipeline 的构建与实战技巧。

1. 什么是 Pipeline？为什么需要它？

Pipeline 是一个将多个数据变换步骤和最终估计器（模型）顺序组合的工具。当你调用 fit 时，它会按照顺序对数据执行所有变换，最终用变换后的数据训练模型；调用 predict 时，同样依次变换数据，再用训练好的模型进行预测。

核心优势：

防止数据泄漏：交叉验证中，所有变换（如标准化）仅拟合在训练集上，避免测试集信息混入。
代码简洁：将多个步骤封装为一个对象，对外只需调用 fit 和 predict。
可读性与复现性：完整工作流一目了然，便于分享和部署。
协同调参：可以使用网格搜索一次性调整所有步骤的超参数。

2. 构建你的第一个 Pipeline：标准化 + 逻辑回归

我们从一个最简单的示例开始：对特征进行标准化，然后训练逻辑回归分类器。

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.datasets import load_breast_cancer

# 加载数据
X, y = load_breast_cancer(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 构建 Pipeline
pipe = Pipeline([
    ('scaler', StandardScaler()),      # 步骤1：标准化
    ('clf', LogisticRegression())      # 步骤2：分类器
])

# 训练整个流水线
pipe.fit(X_train, y_train)

# 评估
print(f"测试集准确率: {pipe.score(X_test, y_test):.3f}")

关键点：Pipeline 的步骤列表由 (名称, 估计器对象) 的元组组成。名称用于超参数访问和调试。

3. 包含特征选择的 Pipeline

很多时候，我们希望在特征变换后加入特征选择，例如使用方差阈值或基于模型的特征重要性。

from sklearn.feature_selection import SelectKBest, f_classif

pipe_select = Pipeline([
    ('scaler', StandardScaler()),
    ('selector', SelectKBest(score_func=f_classif, k=10)),  # 选择前10个与目标最相关的特征
    ('clf', LogisticRegression())
])

pipe_select.fit(X_train, y_train)
print(f"选择特征后的准确率: {pipe_select.score(X_test, y_test):.3f}")

4. 处理混合数据类型：ColumnTransformer + Pipeline

现实数据往往包含数值列和类别列，需要不同的预处理方式。ColumnTransformer 允许对不同列应用不同变换，再将其组合，最后嵌入 Pipeline。

import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder
from sklearn.impute import SimpleImputer

# 假设我们有一个包含数值和类别的DataFrame
df = pd.DataFrame({
    'age': [25, 32, 47, 51],
    'city': ['NY', 'LA', 'SF', 'NY'],
    'income': [50000, 72000, 90000, 65000],
    'target': [0, 1, 0, 1]
})
X = df.drop('target', axis=1)
y = df['target']

# 定义列分组
numeric_features = ['age', 'income']
categorical_features = ['city']

# 数值列处理：缺失值填充 + 标准化
numeric_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

# 类别列处理：缺失值填充 + 独热编码
categorical_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),
    ('onehot', OneHotEncoder(handle_unknown='ignore'))
])

# 使用 ColumnTransformer 组合
preprocessor = ColumnTransformer(
    transformers=[
        ('num', numeric_transformer, numeric_features),
        ('cat', categorical_transformer, categorical_features)
    ])

# 最终 Pipeline
full_pipe = Pipeline([
    ('preprocessor', preprocessor),
    ('clf', LogisticRegression())
])

full_pipe.fit(X, y)
# 注意：ColumnTransformer 默认将输出转为密集数组，若需稀疏矩阵可设置 sparse_threshold=0

5. 在 Pipeline 上进行超参数调优

Pipeline 的最大优势之一，就是能够对内部任何步骤的超参数进行交叉验证和网格搜索。访问参数时使用 步骤名__参数名 双下划线语法。

from sklearn.model_selection import GridSearchCV

# 沿用上面的 full_pipe
param_grid = {
    'preprocessor__num__imputer__strategy': ['mean', 'median'],   # 数值管道的填充策略
    'clf__C': [0.1, 1.0, 10.0],                                    # 逻辑回归的正则化强度
    'clf__penalty': ['l1', 'l2']
}

grid_search = GridSearchCV(full_pipe, param_grid, cv=5, scoring='accuracy')
grid_search.fit(X_train, y_train)   # 实际情况下使用分割好的数据

print(f"最佳参数: {grid_search.best_params_}")
print(f"最佳 CV 得分: {grid_search.best_score_:.3f}")

注意：自定义交叉验证时，务必使用 Pipeline 作为估算器，以保证每次折叠中重新拟合变换器。

6. 高级技巧：自定义转换器

当 scikit-learn 内置的转换器不满足需求时，可以通过继承 BaseEstimator 和 TransformerMixin 创建自定义转换器，并直接放入 Pipeline。

from sklearn.base import BaseEstimator, TransformerMixin
import numpy as np

class LogTransformer(BaseEstimator, TransformerMixin):
    """对指定列进行对数变换 (log1p)"""
    def __init__(self, columns=None):
        self.columns = columns

    def fit(self, X, y=None):
        # 如果 columns 未指定，则默认对所有数值列应用
        if self.columns is None:
            self.columns = X.select_dtypes(include=np.number).columns.tolist()
        return self

    def transform(self, X):
        X = X.copy()
        for col in self.columns:
            X[col] = np.log1p(X[col])
        return X

# 使用自定义转换器
custom_pipe = Pipeline([
    ('log', LogTransformer(columns=['income'])),
    ('scaler', StandardScaler()),
    ('clf', LogisticRegression())
])

7. 完整端到端示例：分类问题实战

将上述知识点整合，处理一个包含数值和类别的 Titanic 数据集的简化版。

import pandas as pd
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression

# 加载数据 (实际可用 seaborn 的 titanic)
# 此处模拟一个简化版
df = pd.DataFrame({
    'pclass': [1, 3, 2, 1, 3],
    'sex': ['male', 'female', 'female', 'male', 'female'],
    'age': [22.0, 38.0, 26.0, 35.0, 28.0],
    'fare': [7.25, 71.2833, 7.925, 53.1, 8.05],
    'embarked': ['S', 'C', 'S', 'S', 'Q'],
    'survived': [0, 1, 1, 0, 1]
})
X = df.drop('survived', axis=1)
y = df['survived']

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0)

# 列分组
numeric_features = ['age', 'fare']
categorical_features = ['pclass', 'sex', 'embarked']

# 数值管道
num_pipe = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

# 类别管道
cat_pipe = Pipeline([
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('encoder', OneHotEncoder(handle_unknown='ignore'))
])

preprocessor = ColumnTransformer([
    ('num', num_pipe, numeric_features),
    ('cat', cat_pipe, categorical_features)
])

# 完整模型
clf_pipe = Pipeline([
    ('prep', preprocessor),
    ('clf', LogisticRegression(max_iter=1000))
])

# 超参数搜索
param_grid = {
    'prep__num__imputer__strategy': ['mean', 'median'],
    'clf__C': [0.1, 1, 10],
    'clf__penalty': ['l2']
}

grid = GridSearchCV(clf_pipe, param_grid, cv=3, scoring='accuracy')
grid.fit(X_train, y_train)

print("最佳CV分数: {:.3f}".format(grid.best_score_))
print("测试集分数: {:.3f}".format(grid.score(X_test, y_test)))

8. 常见陷阱与最佳实践

数据泄漏的防范：永远不要在 fit 前对整个数据集应用 StandardScaler.fit()，Pipeline 会保证安全。
稀疏性保持：OneHotEncoder 等默认输出稀疏矩阵，如果后续步骤（如某些模型）不支持稀疏输入，可以在 ColumnTransformer 中设置 sparse_threshold=0 强制转为密集数组。
缓存中间变换：若 Pipeline 中包含计算昂贵的步骤，可以设置 memory 参数进行缓存，加速重复运行：
```
from tempfile import mkdtemp
cachedir = mkdtemp()
pipe = Pipeline(steps, memory=cachedir)
```
使用 passthrough 或 drop：在 ColumnTransformer 中，对于不需处理的列可以传递 'passthrough' 或 'drop'，避免丢失信息。
调试与访问步骤：可以通过 pipe.named_steps['scaler'] 直接访问已拟合的步骤对象，查看其属性（如均值、方差）。

9. 总结

构建 Pipeline 是 Scikit-learn 中实现可维护、无数据泄漏的机器学习工作流的核心方法。从基础的标准化+模型，到混合数据类型处理、特征选择、自定义转换器和超参数调优，Pipeline 提供了统一而强大的范式。掌握这些技巧后，你的代码将更干净、更健壮，也能更自信地部署到生产环境。

下一步建议：

尝试将 PCA、PolynomialFeatures 等加入 Pipeline。
使用 make_pipeline 快速创建无需命名的管道（步骤名自动生成）。
结合 FeatureUnion 并行组合多个特征工程模块。

Happy Pipelining！