第60课 - Scikit-learn Pipeline 构建完整工作流
学习目标
完成本课后，你将能够：
- 理解 Pipeline 在机器学习工作流中的核心价值
- 掌握使用 Pipeline 串联数据预处理和模型训练的方法
- 学会创建自定义转换器以适应特定数据处理需求
- 实践构建完整的端到端机器学习流水线
- 了解 Pipeline 在交叉验证和模型选择中的优势
核心概念
什么是 Pipeline？
想象一下你是一个工厂的厂长，需要生产汽车。一辆汽车的生产流程包括：冲压→焊接→涂装→总装。如果每个环节都由不同的人独立完成，可能会出现零件不匹配、效率低下等问题。
Pipeline（流水线） 就像这个工厂的自动化生产线，它将机器学习工作流中的多个步骤串联起来，形成一个有序、自动化的处理链。在机器学习中，一个典型的 Pipeline 包含：
- 数据预处理（如特征缩放、编码分类变量）
- 特征工程（如特征提取、特征选择）
- 模型训练（最终的预测模型）
为什么使用 Pipeline？
- 避免数据泄露：确保预处理参数只从训练数据学习
- 简化代码：将多个步骤封装为一个对象
- 便于部署：整个处理流程可作为一个整体保存和加载
- 简化交叉验证：在交叉验证中自动应用相同的预处理
关键组件
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.svm import SVC
# 创建一个简单的Pipeline
pipe = Pipeline([
('scaler', StandardScaler()), # 第一步：标准化
('pca', PCA(n_components=2)), # 第二步：降维
('svm', SVC()) # 第三步：分类模型
])
代码示例
# 完整Pipeline示例：从数据加载到模型评估
import numpy as np
import pandas as pd
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, classification_report
# 1. 加载和准备数据
# 创建一个模拟的数据集（包含数值型和分类型特征）
np.random.seed(42)
data_size = 1000
# 模拟数据集
data = pd.DataFrame({
'age': np.random.normal(40, 10, data_size), # 数值型特征
'income': np.random.normal(50000, 15000, data_size), # 数值型特征
'education': np.random.choice(['高中', '本科', '硕士', '博士'], data_size), # 分类型特征
'gender': np.random.choice(['男', '女'], data_size), # 分类型特征
'purchased': np.random.randint(0, 2, data_size) # 目标变量
})
# 人为添加一些缺失值
data.loc[data.sample(frac=0.1).index, 'age'] = np.nan
data.loc[data.sample(frac=0.05).index, 'income'] = np.nan
print("数据集形状:", data.shape)
print("\n数据预览:")
print(data.head())
print("\n缺失值统计:")
print(data.isnull().sum())
# 2. 划分训练集和测试集
X = data.drop('purchased', axis=1)
y = data['purchased']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 3. 定义特征列类型
numeric_features = ['age', 'income']
categorical_features = ['education', 'gender']
# 4. 创建预处理流水线
# 对数值型特征：填充缺失值 -> 标准化
numeric_transformer = Pipeline(steps=[
('imputer', SimpleImputer(strategy='mean')), # 用均值填充缺失值
('scaler', StandardScaler()) # 标准化
])
# 对分类型特征：填充缺失值 -> 独热编码
categorical_transformer = Pipeline(steps=[
('imputer', SimpleImputer(strategy='constant', fill_value='缺失')), # 用常量填充
('encoder', OneHotEncoder(handle_unknown='ignore')) # 独热编码
])
# 使用ColumnTransformer组合不同的预处理
preprocessor = ColumnTransformer(
transformers=[
('num', numeric_transformer, numeric_features),
('cat', categorical_transformer, categorical_features)
])
# 5. 创建完整的机器学习Pipeline
full_pipeline = Pipeline(steps=[
('preprocessor', preprocessor), # 数据预处理
('classifier', RandomForestClassifier(n_estimators=100, random_state=42)) # 分类模型
])
# 6. 训练和评估模型
print("\n训练模型...")
full_pipeline.fit(X_train, y_train)
# 在测试集上评估
y_pred = full_pipeline.predict(X_test)
print(f"测试集准确率: {accuracy_score(y_test, y_pred):.4f}")
print("\n分类报告:")
print(classification_report(y_test, y_pred))
# 7. 使用交叉验证评估Pipeline
print("\n使用5折交叉验证...")
cv_scores = cross_val_score(full_pipeline, X, y, cv=5, scoring='accuracy')
print(f"交叉验证准确率: {cv_scores.mean():.4f} ± {cv_scores.std():.4f}")
# 8. 保存和加载Pipeline（在实际部署中很有用）
import joblib
joblib.dump(full_pipeline, 'trained_pipeline.pkl')
loaded_pipeline = joblib.load('trained_pipeline.pkl')
print("\nPipeline已保存并加载，预测结果一致性:",
np.array_equal(y_pred, loaded_pipeline.predict(X_test)))
实践练习
练习1：基础Pipeline构建
创建一个Pipeline，包含以下步骤：
- 数据标准化（StandardScaler）
- 主成分分析（PCA，保留90%方差）
- 逻辑回归分类器
使用sklearn的make_classification生成数据，评估模型性能。
预期输出：
- Pipeline对象打印信息
- 交叉验证准确率（约85%以上）
练习2：自定义转换器
创建一个自定义转换器DataFrameSelector，用于从DataFrame中选择指定列。要求：
- 继承BaseEstimator和TransformerMixin
- 实现fit和transform方法
- 与ColumnTransformer结合使用
预期输出：
- 自定义转换器成功选择指定特征列
- Pipeline正常运行
练习3：高级Pipeline应用
构建一个处理混合数据类型的完整Pipeline：
- 数值特征：缺失值填充→标准化
- 分类特征：缺失值填充→目标编码
- 文本特征：TF-IDF向量化
- 最终使用梯度提升树分类器
（可选：使用sklearn的fetch_20newsgroups数据集）
预期输出：
- 多模态数据处理Pipeline
- 文本分类准确率报告
常见错误
错误1：Pipeline步骤顺序错误
# ❌ 错误：先缩放后填充缺失值
pipe = Pipeline([
('scaler', StandardScaler()), # StandardScaler不能处理NaN
('imputer', SimpleImputer()),
('model', SVC())
])
# ✅ 正确：先填充后缩放
pipe = Pipeline([
('imputer', SimpleImputer()),
('scaler', StandardScaler()),
('model', SVC())
])
错误2：混淆Pipeline和FeatureUnion
# ❌ 混淆概念：试图并行处理同一数据的Pipeline
from sklearn.pipeline import FeatureUnion
pipe = Pipeline([
('features', FeatureUnion([
('pca', PCA(n_components=2)),
('select', SelectKBest(k=3))
])),
('model', SVC())
])
# ✅ 正确用法：使用FeatureUnion并行处理同一数据
pipe = Pipeline([
('features', FeatureUnion([
('pca', PCA(n_components=2)),
('select', SelectKBest(k=3))
])),
('model', SVC())
])
# 注意：FeatureUnion会合并两个转换器的输出特征
错误3：在Pipeline外进行数据分割
# ❌ 错误：先预处理再分割
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X) # 使用全部数据计算参数！
X_train, X_test = train_test_split(X_scaled, test_size=0.2)
# ✅ 正确：Pipeline内部处理确保参数只从训练集学习
pipe = Pipeline([
('scaler', StandardScaler()),
('model', SVC())
])
X_train, X_test = train_test_split(X, test_size=0.2) # 分割原始数据
pipe.fit(X_train, y_train) # 在训练集上学习参数
错误4：忽略ColumnTransformer的剩余处理
# ❌ 忽略未指定的特征
ct = ColumnTransformer([
('num', StandardScaler(), ['age', 'income'])
# education和gender列被丢弃！
])
# ✅ 使用remainder参数处理剩余列
ct = ColumnTransformer([
('num', StandardScaler(), ['age', 'income'])
], remainder='passthrough') # 保留其他列
小结
关键要点回顾
- Pipeline的核心价值：将多个处理步骤封装为单一对象，避免数据泄露，简化工作流
- 基本组成：Pipeline包含一系列(name, transform)元组，最后一步通常是估计器
- ColumnTransformer：用于对不同特征子集应用不同的预处理流水线
- 自定义转换器：通过继承BaseEstimator和TransformerMixin创建自定义处理步骤
- 最佳实践：
- 始终在Pipeline内进行数据预处理
- 确保预处理参数只从训练数据学习
- 使用cross_val_score进行评估时，Pipeline会自动处理预处理
- 保存训练好的Pipeline以便部署
Pipeline的工作流程
原始数据 → [预处理步骤1] → [预处理步骤2] → ... → [模型训练] → 预测
下一步学习
在下一课中，我们将学习如何分析经过Pipeline处理后的特征重要性，理解哪些特征对模型的预测贡献最大。
掌握Pipeline是构建健壮、可复现机器学习系统的关键技能，它能让你的工作流更加专业和可靠！
练习编辑器
rust
Loading...