发布于 由admin

`Scikit-learn`的`流水线`（`Pipeline`）：`自动化`的`特征工程`和`模型训练`。

Scikit-learn 流水线：自动化特征工程与模型训练

各位同学，今天我们来深入探讨 scikit-learn 中一个非常强大且实用的工具：Pipeline（流水线）。它的核心作用在于自动化特征工程和模型训练流程，将多个步骤串联起来，极大地简化了机器学习项目的开发、维护和部署过程。

为什么需要流水线？

在实际的机器学习项目中，通常需要进行一系列的预处理步骤，例如：

  • 数据清洗: 处理缺失值、异常值等。
  • 特征缩放: 将特征缩放到相似的范围，如标准化或归一化。
  • 特征编码: 将类别特征转换为数值特征，如独热编码。
  • 特征选择: 选择最相关的特征，降低维度。
  • 模型训练: 使用处理后的数据训练机器学习模型。

如果没有流水线，我们需要手动地对训练集和测试集分别执行这些步骤，这不仅繁琐且容易出错，还可能导致数据泄露（data leakage）。数据泄露是指在模型训练过程中，不小心使用了测试集的信息，导致模型在训练集上的表现看起来很好，但在测试集上的泛化能力却很差。

流水线很好地解决了这些问题：

  1. 简化流程: 将多个步骤组合成一个单一的对象，简化代码。
  2. 避免数据泄露: 对训练集进行拟合，然后将相同的转换应用于测试集，确保测试集的信息不会影响模型的训练。
  3. 方便模型选择和调优: 可以将不同的预处理步骤和模型组合成不同的流水线，进行交叉验证和参数调优，选择最佳的配置。
  4. 易于部署: 将整个预处理和模型训练流程封装在一个对象中，方便模型的部署和维护。

流水线的组成

Pipeline 对象由一系列的步骤组成，每个步骤都是一个 Transformer（转换器）或 Estimator（估计器）。

  • Transformer: 负责数据转换，如 StandardScaler（标准化）、OneHotEncoder（独热编码）等。Transformer 必须实现 fit 和 transform 方法。
  • Estimator: 负责模型训练和预测，如 LogisticRegression（逻辑回归）、DecisionTreeClassifier（决策树）等。Estimator 必须实现 fit 和 predict 方法。

流水线中，除了最后一个步骤可以是 Estimator 之外，其他步骤都必须是 Transformer。这是因为流水线的目标是先对数据进行转换，然后再使用转换后的数据训练模型。

创建流水线

创建流水线有两种方式：

  1. 使用 make_pipeline 函数: 这种方式更简洁，只需要传入 Transformer 和 Estimator 对象即可。
  2. 使用 Pipeline 类: 这种方式更灵活，可以自定义每个步骤的名称。

下面我们分别来看这两种方式的用法。

1. 使用 make_pipeline 函数

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

# 创建流水线
pipeline = make_pipeline(StandardScaler(), LogisticRegression())

print(pipeline)

输出：

Pipeline(steps=[('standardscaler', StandardScaler()),
                ('logisticregression', LogisticRegression())])

在这个例子中，我们创建了一个包含两个步骤的流水线：StandardScaler 和 LogisticRegression。make_pipeline 函数会自动为每个步骤生成名称，名称为类名的小写形式。

2. 使用 Pipeline 类

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

# 创建流水线
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', LogisticRegression())
])

print(pipeline)

输出：

Pipeline(steps=[('scaler', StandardScaler()), ('classifier', LogisticRegression())])

在这个例子中，我们使用 Pipeline 类创建流水线。我们需要传入一个包含多个元组的列表，每个元组包含两个元素：步骤的名称和 Transformer 或 Estimator 对象。使用 Pipeline 类可以自定义每个步骤的名称，方便后续的参数调优。

使用流水线

创建流水线后，我们可以像使用单个 Transformer 或 Estimator 一样使用它。流水线对象拥有 fit、transform、predict 等方法。

from sklearn.model_selection import train_test_split
from sklearn.datasets import load_iris
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score

# 加载数据集
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)

# 创建流水线
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', LogisticRegression())
])

# 训练模型
pipeline.fit(X_train, y_train)

# 预测
y_pred = pipeline.predict(X_test)

# 评估模型
accuracy = accuracy_score(y_test, y_pred)
print(f"Accuracy: {accuracy}")

输出：

Accuracy: 1.0

在这个例子中，我们首先加载了鸢尾花数据集，然后将其划分为训练集和测试集。接着，我们创建了一个包含 StandardScaler 和 LogisticRegression 的流水线。我们使用 fit 方法在训练集上训练流水线，然后使用 predict 方法在测试集上进行预测。最后，我们使用 accuracy_score 函数评估模型的性能。

注意，我们只需要在训练集上调用 fit 方法，流水线会自动将 StandardScaler 应用于训练集，然后使用转换后的数据训练 LogisticRegression 模型。在测试集上，我们只需要调用 predict 方法，流水线会自动将 StandardScaler 应用于测试集，然后使用训练好的 LogisticRegression 模型进行预测。这避免了手动对测试集进行预处理的麻烦，并且确保了数据的一致性，防止了数据泄露。

流水线的优势：避免数据泄露

让我们通过一个具体的例子来说明流水线如何避免数据泄露。假设我们需要对一个包含缺失值的数据集进行处理，并使用逻辑回归模型进行分类。

import numpy as np
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

# 创建包含缺失值的数据集
X = np.array([
    [1, 2, np.nan],
    [3, np.nan, 5],
    [4, 5, 6],
    [7, 8, 9],
    [np.nan, 10, 11]
])
y = np.array([0, 1, 0, 1, 0])

# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)

# 不使用流水线，手动处理缺失值和缩放
imputer = SimpleImputer(strategy='mean')
X_train_imputed = imputer.fit_transform(X_train)
X_test_imputed = imputer.transform(X_test)

scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train_imputed)
X_test_scaled = scaler.transform(X_test_imputed)

model = LogisticRegression()
model.fit(X_train_scaled, y_train)
y_pred = model.predict(X_test_scaled)
accuracy = accuracy_score(y_test, y_pred)
print(f"Accuracy (without pipeline): {accuracy}")

# 使用流水线处理缺失值和缩放
from sklearn.pipeline import Pipeline

pipeline = Pipeline([
    ('imputer', SimpleImputer(strategy='mean')),
    ('scaler', StandardScaler()),
    ('classifier', LogisticRegression())
])

pipeline.fit(X_train, y_train)
y_pred = pipeline.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
print(f"Accuracy (with pipeline): {accuracy}")

在这个例子中，我们创建了一个包含缺失值的数据集。如果不使用流水线，我们需要先使用 SimpleImputer 填充训练集和测试集中的缺失值，然后使用 StandardScaler 对训练集和测试集进行缩放。注意，我们需要分别在训练集上 fit SimpleImputer 和 StandardScaler，然后在测试集上使用 transform 方法。

如果我们在填充缺失值或缩放时，不小心使用了测试集的信息，就会导致数据泄露。例如，如果我们使用整个数据集（包括训练集和测试集）来 fit SimpleImputer，那么模型在训练时就会知道测试集中的缺失值分布，这会导致模型在测试集上的表现看起来更好，但实际上泛化能力却很差。

使用流水线可以避免这种问题。当我们使用流水线时，只需要在训练集上调用 fit 方法，流水线会自动将 SimpleImputer 和 StandardScaler 应用于训练集，然后使用转换后的数据训练 LogisticRegression 模型。在测试集上，我们只需要调用 predict 方法，流水线会自动将 SimpleImputer 和 StandardScaler 应用于测试集，然后使用训练好的 LogisticRegression 模型进行预测。这样可以确保测试集的信息不会影响模型的训练，避免了数据泄露。

流水线与模型选择和调优

流水线不仅可以简化流程和避免数据泄露，还可以方便模型选择和调优。我们可以将不同的预处理步骤和模型组合成不同的流水线，然后使用交叉验证和参数调优选择最佳的配置。

from sklearn.model_selection import GridSearchCV
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.svm import SVC
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split

# 加载数据集
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)

# 定义参数网格
param_grid = [
    {
        'classifier': [LogisticRegression()],
        'classifier__penalty': ['l1', 'l2'],
        'classifier__C': [0.1, 1, 10],
        'classifier__solver': ['liblinear']
    },
    {
        'classifier': [SVC()],
        'classifier__C': [0.1, 1, 10],
        'classifier__kernel': ['linear', 'rbf']
    }
]

# 创建流水线
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', LogisticRegression())  # 初始模型，会被 GridSearchCV 替换
])

# 创建 GridSearchCV 对象
grid_search = GridSearchCV(pipeline, param_grid, cv=5, scoring='accuracy')

# 训练模型
grid_search.fit(X_train, y_train)

# 打印最佳参数和得分
print(f"Best parameters: {grid_search.best_params_}")
print(f"Best score: {grid_search.best_score_}")

# 在测试集上评估模型
accuracy = grid_search.score(X_test, y_test)
print(f"Accuracy on test set: {accuracy}")

在这个例子中，我们首先定义了一个参数网格 param_grid，它包含了我们想要调优的参数。我们想要比较 LogisticRegression 和 SVC 两种模型，并对它们的参数进行调优。

然后，我们创建了一个包含 StandardScaler 和 LogisticRegression 的流水线。注意，这里的 LogisticRegression 只是一个占位符，它会被 GridSearchCV 替换为 param_grid 中定义的模型。

接着，我们创建了一个 GridSearchCV 对象，它会自动将不同的预处理步骤和模型组合成不同的流水线，并使用交叉验证选择最佳的配置。

最后，我们使用 fit 方法在训练集上训练 GridSearchCV 对象，然后使用 best_params_ 和 best_score_ 属性获取最佳参数和得分。我们还可以使用 score 方法在测试集上评估模型的性能。

通过使用流水线和 GridSearchCV，我们可以方便地进行模型选择和调优，找到最佳的配置，提高模型的性能。

自定义 Transformer

有时候，scikit-learn 提供的 Transformer 无法满足我们的需求，我们需要自定义 Transformer。自定义 Transformer 必须继承 TransformerMixin 和 BaseEstimator 类，并实现 fit 和 transform 方法。

from sklearn.base import BaseEstimator, TransformerMixin

class CustomTransformer(BaseEstimator, TransformerMixin):
    def __init__(self, feature_name):
        self.feature_name = feature_name

    def fit(self, X, y=None):
        return self

    def transform(self, X):
        # 在这里实现自定义的转换逻辑
        # 例如，提取指定特征的平方
        return X[self.feature_name] ** 2

在这个例子中，我们定义了一个名为 CustomTransformer 的自定义 Transformer，它可以提取指定特征的平方。fit 方法通常不需要做任何事情，只需要返回 self 即可。transform 方法实现了自定义的转换逻辑。

import pandas as pd
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

# 创建 DataFrame
data = {'feature1': [1, 2, 3, 4, 5],
        'feature2': [6, 7, 8, 9, 10]}
df = pd.DataFrame(data)

# 创建流水线
pipeline = Pipeline([
    ('custom_transformer', CustomTransformer(feature_name='feature1')),
    ('scaler', StandardScaler())
])

# 转换数据
transformed_data = pipeline.fit_transform(df)

print(transformed_data)

在这个例子中，我们首先创建了一个 DataFrame，然后创建了一个包含 CustomTransformer 和 StandardScaler 的流水线。我们使用 fit_transform 方法将流水线应用于 DataFrame，得到转换后的数据。

ColumnTransformer：处理不同类型的特征

在实际的项目中，数据集通常包含不同类型的特征，例如数值特征、类别特征、文本特征等。我们需要对不同类型的特征进行不同的预处理。ColumnTransformer 可以帮助我们实现这个目标。

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
import pandas as pd

# 创建 DataFrame
data = {
    'numerical_feature': [1, 2, 3, 4, 5],
    'categorical_feature': ['A', 'B', 'A', 'C', 'B'],
    'target': [0, 1, 0, 1, 0]
}
df = pd.DataFrame(data)

# 划分训练集和测试集
X = df.drop('target', axis=1)
y = df['target']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)

# 定义 ColumnTransformer
numerical_features = ['numerical_feature']
categorical_features = ['categorical_feature']

preprocessor = ColumnTransformer(
    transformers=[
        ('numerical', StandardScaler(), numerical_features),
        ('categorical', OneHotEncoder(), categorical_features)
    ])

# 创建流水线
pipeline = Pipeline([
    ('preprocessor', preprocessor),
    ('classifier', LogisticRegression())
])

# 训练模型
pipeline.fit(X_train, y_train)

# 预测
y_pred = pipeline.predict(X_test)

# 评估模型
from sklearn.metrics import accuracy_score
accuracy = accuracy_score(y_test, y_pred)
print(f"Accuracy: {accuracy}")

在这个例子中，我们首先创建了一个包含数值特征和类别特征的 DataFrame。然后，我们定义了一个 ColumnTransformer 对象，它指定了对不同类型的特征进行不同的预处理。对于数值特征，我们使用 StandardScaler 进行缩放；对于类别特征，我们使用 OneHotEncoder 进行独热编码。

接着，我们创建了一个包含 ColumnTransformer 和 LogisticRegression 的流水线。我们使用 fit 方法在训练集上训练流水线，然后使用 predict 方法在测试集上进行预测。

ColumnTransformer 可以方便地处理不同类型的特征，提高模型的性能。

流水线在实际项目中的应用

流水线在实际的机器学习项目中应用非常广泛。以下是一些常见的应用场景：

  • 自然语言处理: 可以使用流水线来处理文本数据，例如分词、词干提取、TF-IDF 向量化等。
  • 图像识别: 可以使用流水线来处理图像数据，例如图像缩放、颜色空间转换、特征提取等。
  • 金融风控: 可以使用流水线来处理金融数据，例如缺失值处理、异常值检测、特征选择等。
  • 推荐系统: 可以使用流水线来处理用户行为数据，例如数据清洗、特征工程、模型训练等。

流水线可以帮助我们简化开发流程，提高代码的可维护性，并且避免数据泄露，是机器学习项目中不可或缺的工具。

流水线让流程更清晰，训练更安全

本次讲座我们深入了解了 scikit-learn 中的 Pipeline 的概念、组成和使用方法。它将特征工程和模型训练自动化，简化了机器学习项目的开发、维护和部署。通过流水线，我们能避免数据泄露，提高模型泛化能力，并方便地进行模型选择和参数调优，是机器学习工程师的重要工具。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注