【DAY27】pipeline管道

@浙大疏锦行

一、先明确核心概念对应关系

• 转化器：带fit()+transform()的对象（如StandardScaler），负责数据预处理；
• 估计器：带fit()+predict()的对象（如LogisticRegression），负责模型训练 / 预测；
• ColumnTransformer：封装 “不同列用不同转化器” 的逻辑（解决多类型特征的预处理）；
• Pipeline：串联 “预处理（含 ColumnTransformer）+ 模型（估计器）” 的流程容器。

二、通用机器学习 Pipeline 的逻辑顺序

1. 数据加载与初步校验

   • 加载数据集（如 CSV），检查基础信息：数据形状、数据类型、缺失值 / 异常值分布。

2. 特征与标签分离

   • 将数据集拆分为特征矩阵（X）和目标变量（y）（监督学习必备步骤）。

3. 预处理逻辑封装（转化器 + ColumnTransformer）

   • 针对不同特征列，选择对应转化器：
     • 数值列：用StandardScaler（标准化）、MinMaxScaler（归一化）等转化器；
     • 分类型列：用OneHotEncoder（独热编码）、OrdinalEncoder（标签编码）等转化器；
   • 用ColumnTransformer将 “列→转化器” 的映射打包（作为 Pipeline 的预处理步骤）。

4. 构建 Pipeline（串联预处理 + 模型）

   • Pipeline 的步骤格式：[("预处理步骤名", ColumnTransformer对象), ("模型步骤名", 估计器对象)]
   • 示例：Pipeline([("preprocess", col_transformer), ("model", LogisticRegression())])

5. Pipeline 训练（自动串联流程）

   • 对 Pipeline 调用fit(X, y)：自动执行 “预处理（fit+transform）→ 模型训练（fit）”，避免数据泄露。

6. 模型预测与评估

   • 调用 Pipeline 的predict(X)得到结果，用对应指标（分类用准确率、回归用 MSE）评估模型效果

# ===================== 通用机器学习Pipeline完整示例（泰坦尼克生存预测） ===================== # 核心：覆盖数据加载、特征拆分、多类型预处理、Pipeline构建、训练评估、参数调优全流程 # 1. 导入所有核心依赖库 import pandas as pd import numpy as np from sklearn.datasets import fetch_openml from sklearn.model_selection import train_test_split, GridSearchCV from sklearn.preprocessing import StandardScaler, OneHotEncoder from sklearn.impute import SimpleImputer from sklearn.compose import ColumnTransformer from sklearn.pipeline import Pipeline from sklearn.linear_model import LogisticRegression from sklearn.metrics import accuracy_score, classification_report # 2. 加载并预处理数据集（泰坦尼克：含数值/分类特征+缺失值，贴近真实场景） def load_and_split_data(): """加载数据集并划分训练/测试集""" # 加载泰坦尼克公开数据集 titanic = fetch_openml("titanic", version=1, as_frame=True, parser="pandas") df = titanic.frame # 特征/标签分离（简化特征，聚焦核心逻辑） features = ["age", "fare", "pclass", "sex", "embarked"] # 数值：age/fare；分类：pclass/sex/embarked X = df[features] y = df["survived"].astype(int) # 标签：1=生存，0=未生存 # 划分训练集/测试集（stratify保证标签分布一致，避免数据泄露） X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) return X_train, X_test, y_train, y_test # 3. 构建预处理管道（区分数值/分类特征） def build_preprocessor(): """构建ColumnTransformer，整合不同特征的预处理逻辑""" # 特征分组 numeric_features = ["age", "fare"] categorical_features = ["pclass", "sex", "embarked"] # 数值特征预处理：填充缺失值（中位数）+ 标准化 numeric_transformer = Pipeline(steps=[ ("imputer", SimpleImputer(strategy="median")), ("scaler", StandardScaler()) ]) # 分类特征预处理：填充缺失值（众数）+ 独热编码（忽略未知类别） categorical_transformer = Pipeline(steps=[ ("imputer", SimpleImputer(strategy="most_frequent")), ("onehot", OneHotEncoder(handle_unknown="ignore")) ]) # 整合两类特征的预处理 preprocessor = ColumnTransformer( transformers=[ ("num", numeric_transformer, numeric_features), ("cat", categorical_transformer, categorical_features) ] ) return preprocessor # 4. 构建完整Pipeline（预处理 + 模型）+ 训练评估 + 参数调优 def main(): # 加载数据 X_train, X_test, y_train, y_test = load_and_split_data() # 构建预处理组件 preprocessor = build_preprocessor() # 构建完整Pipeline（预处理 + 逻辑回归模型） full_pipeline = Pipeline(steps=[ ("preprocessor", preprocessor), ("classifier", LogisticRegression(max_iter=1000, random_state=42)) ]) # ========== 基础训练与评估 ========== print("===== 基础模型训练与评估 =====") # 训练（自动执行：预处理fit+transform → 模型fit） full_pipeline.fit(X_train, y_train) # 预测（自动执行：预处理transform → 模型predict） 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)) # ========== 管道参数调优（GridSearchCV） ========== print("\n===== 开始Pipeline参数调优 =====") # 定义待调优参数（格式：步骤名__参数名） param_grid = { "preprocessor__num__imputer__strategy": ["mean", "median"], # 数值缺失值填充策略 "classifier__C": [0.1, 1.0, 10.0] # 逻辑回归正则化系数 } # 网格搜索（5折交叉验证，按准确率评分） grid_search = GridSearchCV( full_pipeline, param_grid, cv=5, scoring="accuracy", n_jobs=-1, random_state=42 ) grid_search.fit(X_train, y_train) # 输出调优结果 print(f"\n最优参数组合：{grid_search.best_params_}") print(f"交叉验证最优准确率：{grid_search.best_score_:.4f}") # 用最优模型评估测试集 best_y_pred = grid_search.predict(X_test) print(f"\n调优后测试集准确率：{accuracy_score(y_test, best_y_pred):.4f}") print("\n调优后分类报告：") print(classification_report(y_test, best_y_pred)) # 执行主函数 if __name__ == "__main__": main()