2.6 管道 (Pipeline)

文档摘要

2.6 管道 (Pipeline) Scikit-learn 管道 (Pipeline) 详解与代码实践 2.6 管道 (Pipeline) 在机器学习的实际应用中，数据预处理和模型训练往往不是孤立的步骤，而是一个流程化的过程。例如，在训练模型之前，我们可能需要对数据进行标准化、特征编码、降维等多种预处理操作。如果这些步骤是分散进行的，不仅代码冗余，而且容易出错，尤其是在交叉验证和模型部署等场景下。 Scikit-learn 的管道 (Pipeline) 组件正是为了解决这个问题而生的。它允许我们将多个数据预处理步骤和最终的估计器（estimator，例如分类器、回归器）串联起来，形成一个清晰、易于管理和高效的机器学习工作流。

2.6 管道 (Pipeline)

Scikit-learn 管道 (Pipeline) 详解与代码实践

2.6 管道 (Pipeline)

在机器学习的实际应用中，数据预处理和模型训练往往不是孤立的步骤，而是一个流程化的过程。例如，在训练模型之前，我们可能需要对数据进行标准化、特征编码、降维等多种预处理操作。如果这些步骤是分散进行的，不仅代码冗余，而且容易出错，尤其是在交叉验证和模型部署等场景下。

Scikit-learn 的 管道 (Pipeline) 组件正是为了解决这个问题而生的。它允许我们将多个数据预处理步骤和最终的估计器（estimator，例如分类器、回归器）串联起来，形成一个清晰、易于管理和高效的机器学习工作流。管道可以将多个 Scikit-learn 转换器 (transformer) 和一个估计器 (estimator) 组合成一个单一的复合估计器。

2.6.1 管道的核心概念

管道 (Pipeline) 本质上是一个顺序执行的步骤列表，其中除了最后一个步骤必须是估计器 (estimator) (例如 LogisticRegression, RandomForestClassifier, SVR 等)，其余步骤必须是转换器 (transformer) (例如 StandardScaler, PCA, OneHotEncoder 等)。

转换器 (Transformer): 转换器是 Scikit-learn 中用于数据预处理的组件，它实现了 fit 和 transform 方法 (或者 fit_transform 方法)。fit 方法用于学习数据的参数 (例如 StandardScaler 的均值和标准差，PCA 的主成分)，transform 方法使用学习到的参数来转换数据。
估计器 (Estimator): 估计器是 Scikit-learn 中用于模型训练和预测的组件，它实现了 fit 和 predict 方法 (或者 fit, predict, score 等方法)。fit 方法用于使用训练数据训练模型，predict 方法用于使用训练好的模型进行预测。

管道的工作流程如下:

fit(X, y) 方法: 当调用管道的 fit 方法时，它会依次调用管道中每个步骤的 fit_transform 方法 (除了最后一个步骤，只调用 fit 方法)。
- 对于每个转换器，fit_transform 方法会先调用 fit(X, y) 学习数据参数，然后调用 transform(X) 使用学习到的参数转换数据，并将转换后的数据传递给管道的下一个步骤。
- 对于最后一个估计器，只调用 fit(X, y) 方法进行模型训练，输入数据是经过前面所有转换器处理后的数据。
transform(X) 方法: 当调用管道的 transform 方法时，它会依次调用管道中每个转换器的 transform(X) 方法，将数据按照管道中定义的顺序进行转换。
predict(X) 方法: 当调用管道的 predict 方法时，数据会先经过管道中所有转换器的 transform(X) 方法进行预处理，然后将预处理后的数据传递给最后一个估计器的 predict(X) 方法进行预测。

管道的步骤:

管道的步骤是一个包含名称和转换器/估计器对的列表。例如：


from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
pipeline = Pipeline([
    ('scaler', StandardScaler()),  # 步骤1: 标准化，名称为 'scaler'，转换器为 StandardScaler()
    ('classifier', LogisticRegression()) # 步骤2: 分类，名称为 'classifier'，估计器为 LogisticRegression()
])

在这个例子中，管道包含两个步骤：

'scaler': 使用 StandardScaler 进行数据标准化。
'classifier': 使用 LogisticRegression 进行逻辑回归分类。

2.6.2 管道的优势

使用管道可以带来诸多优势，使得机器学习工作流更加清晰、高效和可靠：

代码组织与可读性: 管道将多个步骤整合到一个对象中，使得代码更加结构化，易于理解和维护。避免了手动管理中间数据转换的复杂性，减少了代码冗余。
防止数据泄露 (Data Leakage): 在交叉验证等场景中，如果不使用管道，很容易在预处理阶段发生数据泄露。例如，在交叉验证的每个折叠中，应该只使用训练集数据来拟合预处理器，然后将预处理器应用于训练集和测试集。管道可以确保预处理步骤只在训练集上拟合，然后应用于验证集或测试集，避免信息泄露，得到更可靠的评估结果。
简化模型参数调优: 当使用 GridSearchCV 或 RandomizedSearchCV 进行超参数调优时，可以直接对整个管道进行调优，包括预处理步骤的参数和模型参数。这使得参数调优更加方便和全面。
方便模型部署和复用: 训练好的管道可以作为一个整体保存和加载，方便模型的部署和复用。部署时，只需要加载管道对象，即可对新的输入数据进行预处理和预测，无需手动重复预处理步骤。
减少错误: 手动进行数据预处理和模型训练容易出错，例如忘记应用某个预处理步骤，或者预处理步骤的顺序错误。管道可以自动化整个流程，减少人为错误。

2.6.3 代码实践：构建简单的管道

我们首先通过一个简单的例子来演示如何构建和使用管道。我们将使用经典的鸢尾花 (iris) 数据集，进行数据标准化和逻辑回归分类。


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_iris
from sklearn.metrics import accuracy_score
# 1. 加载数据集
iris = load_iris()
X, y = iris.data, iris.target
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
# 2. 构建管道
pipeline = Pipeline([
    ('scaler', StandardScaler()),  # 标准化步骤
    ('classifier', LogisticRegression()) # 逻辑回归分类器步骤
])
# 3. 训练管道
pipeline.fit(X_train, y_train)
# 4. 预测
y_pred = pipeline.predict(X_test)
# 5. 评估模型
accuracy = accuracy_score(y_test, y_pred)
print(f"模型准确率: {accuracy:.2f}")

代码详解:

加载数据集: 使用 load_iris() 加载鸢尾花数据集，并划分为训练集和测试集。
构建管道: 使用 Pipeline 类创建一个管道对象。
- Pipeline([]) 接受一个步骤列表作为参数。
- 每个步骤是一个元组 ('名称', 转换器/估计器)。
- 这里我们定义了两个步骤:
  - 'scaler': 使用 StandardScaler() 进行数据标准化。
  - 'classifier': 使用 LogisticRegression() 作为分类器。
训练管道: 调用 pipeline.fit(X_train, y_train) 训练管道。
- 管道会先调用 StandardScaler().fit_transform(X_train, y_train) 对训练集数据进行标准化。
- 然后将标准化后的训练集数据传递给 LogisticRegression().fit() 进行模型训练。
预测: 调用 pipeline.predict(X_test) 进行预测。
- 管道会先调用 StandardScaler().transform(X_test) 对测试集数据进行标准化 (使用训练集上学到的参数)。
- 然后将标准化后的测试集数据传递给 LogisticRegression().predict() 进行预测。
评估模型: 使用 accuracy_score 计算模型在测试集上的准确率。

运行结果:


模型准确率: 1.00

可以看到，使用管道可以简洁地完成数据预处理和模型训练，并得到较高的分类准确率。

2.6.4 代码实践：使用 ColumnTransformer 处理不同类型特征

在实际应用中，数据集往往包含不同类型的特征，例如数值型特征、类别型特征等。我们需要对不同类型的特征进行不同的预处理。 ColumnTransformer 可以帮助我们实现对不同列应用不同的转换器，并将其与管道结合使用。


from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
import pandas as pd
# 1. 创建示例数据 (DataFrame)
data = {
    'age': [25, 30, 35, 40, 45, 50],
    'city': ['Beijing', 'Shanghai', 'Guangzhou', 'Beijing', 'Shanghai', 'Shenzhen'],
    'income': [50000, 60000, 70000, 80000, 90000, 100000],
    'bought_product': [0, 1, 1, 0, 1, 1] # 0: No, 1: Yes
}
df = pd.DataFrame(data)
# 划分特征和目标变量
X = df[['age', 'city', 'income']]
y = df['bought_product']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
# 2. 定义 ColumnTransformer
preprocessor = ColumnTransformer(
    transformers=[
        ('num', StandardScaler(), ['age', 'income']), # 对数值型特征 'age' 和 'income' 进行标准化
        ('cat', OneHotEncoder(), ['city']) # 对类别型特征 'city' 进行独热编码
    ])
# 3. 构建包含 ColumnTransformer 的管道
pipeline = Pipeline([
    ('preprocessor', preprocessor), # 预处理步骤
    ('classifier', LogisticRegression()) # 逻辑回归分类器步骤
])
# 4. 训练管道
pipeline.fit(X_train, y_train)
# 5. 预测
y_pred = pipeline.predict(X_test)
# 6. 评估模型 (这里示例数据较小，评估意义不大，仅作演示)
accuracy = accuracy_score(y_test, y_pred)
print(f"模型准确率: {accuracy:.2f}")

代码详解:

创建示例数据: 我们创建了一个包含数值型特征 ('age', 'income') 和类别型特征 ('city') 的 DataFrame。
定义 ColumnTransformer: 使用 ColumnTransformer 定义预处理器。
- transformers 参数是一个列表，每个元素是一个元组 ('名称', 转换器, 列名列表)。
- 'num': 对列 'age' 和 'income' 应用 StandardScaler 进行标准化。
- 'cat': 对列 'city' 应用 OneHotEncoder 进行独热编码。
构建管道: 构建管道时，将 ColumnTransformer 对象作为第一个步骤 ('preprocessor') 加入管道。
训练、预测和评估: 训练、预测和评估步骤与之前的例子相同，管道会自动处理 ColumnTransformer 的应用。

运行结果 (示例数据，仅供参考):


模型准确率: 1.00

ColumnTransformer 使得我们可以灵活地对不同类型的特征应用不同的预处理方法，并将其 seamlessly 集成到管道中，处理复杂的数据预处理流程。

2.6.5 代码实践：自定义转换器 (Custom Transformer)

除了使用 Scikit-learn 内置的转换器，我们还可以根据实际需求创建自定义的转换器，并将其加入管道。自定义转换器需要继承 TransformerMixin 和 BaseEstimator 类，并实现 fit 和 transform 方法。


from sklearn.base import BaseEstimator, TransformerMixin
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_iris
from sklearn.metrics import accuracy_score
import numpy as np
# 1. 自定义特征工程转换器 (示例: 计算特征的平方)
class FeatureSquareTransformer(BaseEstimator, TransformerMixin):
    def __init__(self, feature_indices=None):
        self.feature_indices = feature_indices # 指定要计算平方的特征列索引，默认为所有列
    def fit(self, X, y=None):
        return self # fit 方法通常不需要做任何操作，返回 self 即可
    def transform(self, X):
        if self.feature_indices is None:
            X_squared = X ** 2 # 对所有列计算平方
        else:
            X_squared = X[:, self.feature_indices] ** 2 # 对指定列计算平方
        return np.concatenate((X, X_squared), axis=1) # 将原始特征和平方特征拼接
# 2. 加载数据集
iris = load_iris()
X, y = iris.data, iris.target
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
# 3. 构建包含自定义转换器的管道
pipeline = Pipeline([
    ('scaler', StandardScaler()), # 标准化步骤
    ('feature_square', FeatureSquareTransformer(feature_indices=[0, 2])), # 自定义特征工程步骤 (对第 1 列和第 3 列特征计算平方)
    ('classifier', LogisticRegression()) # 逻辑回归分类器步骤
])
# 4. 训练管道
pipeline.fit(X_train, y_train)
# 5. 预测
y_pred = pipeline.predict(X_test)
# 6. 评估模型
accuracy = accuracy_score(y_test, y_pred)
print(f"模型准确率: {accuracy:.2f}")

代码详解:

自定义转换器 FeatureSquareTransformer:
- 继承 BaseEstimator 和 TransformerMixin。
- __init__ 方法: 初始化参数 feature_indices，用于指定要计算平方的特征列索引。
- fit 方法: 通常不需要做任何操作，直接返回 self。
- transform 方法:
  - 根据 feature_indices 计算指定列或所有列的平方。
  - 使用 np.concatenate 将原始特征和平方特征拼接在一起。
构建管道: 在管道中添加自定义转换器 FeatureSquareTransformer 作为一个步骤 ('feature_square')。
训练、预测和评估: 训练、预测和评估步骤与之前的例子相同，管道会自动调用自定义转换器的 fit 和 transform 方法。

自定义转换器使得我们可以灵活地扩展 Scikit-learn 的功能，实现各种复杂的数据预处理和特征工程操作，并将其整合到管道中。

2.6.6 代码实践：管道与 GridSearchCV 结合进行超参数调优

管道与 GridSearchCV 或 RandomizedSearchCV 结合使用，可以方便地对整个管道 (包括预处理步骤和模型) 进行超参数调优。


from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import GridSearchCV, train_test_split
from sklearn.datasets import load_iris
# 1. 加载数据集
iris = load_iris()
X, y = iris.data, iris.target
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
# 2. 构建管道
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', LogisticRegression())
])
# 3. 定义参数网格 (注意参数名称的格式: '步骤名称__参数名称')
param_grid = {
    'scaler__with_mean': [True, False], # StandardScaler 的参数
    'classifier__C': [0.1, 1.0, 10.0], # LogisticRegression 的参数
    'classifier__penalty': ['l1', 'l2'], # LogisticRegression 的参数
    'classifier__solver': ['liblinear'] # LogisticRegression 的参数
}
# 4. 使用 GridSearchCV 进行超参数调优
grid_search = GridSearchCV(pipeline, param_grid, cv=5) # 5 折交叉验证
grid_search.fit(X_train, y_train)
# 5. 获取最佳模型和参数
best_pipeline = grid_search.best_estimator_
best_params = grid_search.best_params_
print(f"最佳参数: {best_params}")
print(f"最佳模型在测试集上的准确率: {best_pipeline.score(X_test, y_test):.2f}")

代码详解:

构建管道: 与之前例子相同。
定义参数网格 param_grid: 这是关键步骤。
- 参数名称格式: '步骤名称__参数名称'。例如，要调优 StandardScaler 的 with_mean 参数，参数名称为 'scaler__with_mean'。要调优 LogisticRegression 的 C 参数，参数名称为 'classifier__C'。
- 参数网格是一个字典，键是参数名称，值是参数的候选值列表。
使用 GridSearchCV: 使用 GridSearchCV 对管道进行超参数调优。
- 将管道对象 pipeline 和参数网格 param_grid 传递给 GridSearchCV。
- cv=5 表示使用 5 折交叉验证。
- 调用 grid_search.fit(X_train, y_train) 进行训练和调优。
获取最佳模型和参数:
- grid_search.best_estimator_ 获取在交叉验证中性能最佳的管道对象。
- grid_search.best_params_ 获取最佳参数组合。
- 使用 best_pipeline.score(X_test, y_test) 评估最佳模型在测试集上的性能。

运行结果 (示例):


最佳参数: {'classifier__C': 1.0, 'classifier__penalty': 'l1', 'classifier__solver': 'liblinear', 'scaler__with_mean': True}
最佳模型在测试集上的准确率: 1.00

通过 GridSearchCV 与管道结合，我们可以方便地对整个机器学习流程进行自动化超参数调优，找到最佳的预处理和模型参数组合，提升模型性能。

2.6.7 管道的持久化 (Persistence)

训练好的管道可以持久化保存到磁盘，并在需要时加载，方便模型的部署和复用。可以使用 joblib 库进行管道的保存和加载。


from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.datasets import load_iris
from joblib import dump, load
# 1. 加载数据集和构建管道 (省略，与之前例子相同)
iris = load_iris()
X, y = iris.data, iris.target
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', LogisticRegression())
])
pipeline.fit(X, y) # 使用完整数据集训练
# 2. 保存管道到文件
pipeline_filename = 'iris_pipeline.joblib'
dump(pipeline, pipeline_filename)
print(f"管道已保存到文件: {pipeline_filename}")
# 3. 加载管道
loaded_pipeline = load(pipeline_filename)
print("管道已加载")
# 4. 使用加载的管道进行预测
new_data = [[5.1, 3.5, 1.4, 0.2], [6.2, 2.9, 4.3, 1.3]] # 新数据
predictions = loaded_pipeline.predict(new_data)
print(f"新数据预测结果: {predictions}")

代码详解:

保存管道: 使用 dump(pipeline, pipeline_filename) 将训练好的管道 pipeline 保存到文件 iris_pipeline.joblib。
加载管道: 使用 load(pipeline_filename) 从文件 iris_pipeline.joblib 加载管道到 loaded_pipeline 对象。
使用加载的管道预测: 使用加载的管道 loaded_pipeline 对新数据进行预测，与使用原始管道的方式相同。

通过管道的持久化，我们可以轻松地将训练好的模型部署到生产环境，只需加载管道对象，即可对新的输入数据进行预处理和预测，无需重新训练模型或手动进行预处理步骤。

2.6.8 总结

Scikit-learn 管道 (Pipeline) 是一个强大的工具，它能够将多个数据预处理步骤和模型训练步骤整合到一个统一的工作流中，极大地提高了机器学习代码的组织性、可读性和可维护性。管道的优势包括：

代码组织与可读性: 结构化代码，降低复杂度。
防止数据泄露: 确保交叉验证和模型评估的可靠性。
简化参数调优: 方便对整个流程进行超参数搜索。
方便模型部署和复用: 易于保存、加载和部署模型。
减少错误: 自动化流程，减少人为错误。

通过本文的代码实践和内容详解，相信读者已经掌握了 Scikit-learn 管道的基本概念和使用方法。在实际机器学习项目中，强烈建议使用管道来构建模型，提升工作效率和模型质量。掌握管道的使用是成为一名高效的 Scikit-learn 使用者的重要一步。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： https://aiknowledge.cn/collection/176-scikit-learn%E6%9C%BA%E5%99%A8%E5%AD%A6%E4%B9%A0%E5%AE%9E%E6%88%98%E6%95%99%E7%A8%8B

整理：灏天文库整理

由灏天文库平台收录，内容或由平台用户上传，仅供学习交流