day 23 机器学习管道（pipeline）

在机器学习领域，“pipeline” 常被翻译为 “管道” 或 “流水线”，它是机器学习中极为重要的概念。在构建机器学习模型时，通常需按特定顺序对数据执行预处理、特征提取、模型训练以及模型评估等步骤，而使用 “pipeline” 能有效管理这些步骤，构建完整的机器学习流程。

一、转换器（Transformer）

1.定义与功能

转换器是一类特殊的估计器，主要用于数据预处理和特征提取。通过transform 方法能对数据进行各种转换操作。这些操作既包括数据预处理，如归一化、标准化以及缺失值填充等；也涵盖特征提取，例如特征选择、特征组合等。

2.特点

转换器具有无状态性，即它不会存储关于数据的内部状态信息（不存储内参）。它仅依据输入数据学习转换规则（例如函数规律、外部参数），并将这些规则应用于新的数据。这使得转换器可在训练集上学习转换规则，然后直接应用于训练集之外的新数据。

3.常见示例

数据缩放器：像 StandardScaler 和 MinMaxScaler。StandardScaler 通过减去均值并除以标准差来标准化数据，MinMaxScaler 则将数据缩放到指定的最小值和最大值之间。

特征选择器：如 SelectKBest 会根据某种评估指标选择前 K 个最佳特征，PCA（主成分分析）则通过线性变换将高维数据转换为低维数据，同时保留数据的主要特征。

特征提取器：如CountVectorizer 将文本转换为词频矩阵，TF - IDFVectorizer 不仅考虑词频，还会考虑词在文档集合中的稀有程度，从而提取更具代表性的文本特征。

4.示例代码

from sklearn.preprocessing import StandardScaler

# 初始化转换器
scaler = StandardScaler()

# 1. 学习训练数据的缩放规则（计算均值和标准差）,本身不存储数据
scaler.fit(X_train)

# 2. 应用规则到训练数据和测试数据
X_train_scaled = scaler.transform(X_train)
X_test_scaled = scaler.transform(X_test)

# 也可以使用fit_transform一步完成
# X_train_scaled = scaler.fit_transform(X_train)

二、估计器（Estimator）

1.定义与功能

估计器是实现机器学习算法的对象或类。其主要目的是拟合（fit）数据并进行预测（predict）。它是机器学习模型的核心组成部分，通过从数据中学习模式，实现预测任务以及模型评估。

2.特点

估计器具有有状态性，即在训练过程中，它会存储关于数据的状态信息（即内参），这些信息用于预测阶段。估计器通过学习训练数据中的模式和规律来进行预测，因此必须在训练集上进行训练，利用训练得到的模型参数对新数据进行预测。

3.常见示例

分类器：用于将数据分类到不同类别，例如逻辑回归分类器 LogisticRegression、决策树分类器 DecisionTreeClassifier 等。

回归器：主要用来预测连续数值，如线性回归器 LinearRegression，它通过拟合一条直线来预测目标变量的值。

聚类器：旨在将数据划分为不同的簇，例如 KMeans 聚类算法，它将数据点划分为 K 个簇，使得同一簇内的数据点相似度较高。

4.示例代码

from sklearn.linear_model import LinearRegression

# 创建一个回归器
model = LinearRegression()

# 在训练集上训练模型
model.fit(X_train_scaled, y_train)

# 对测试集进行预测
y_pred = model.predict(X_test_scaled)

三、管道（Pipeline）

1.定义与组成

在实际机器学习流程中，通常会先使用转换器对数据进行预处理和特征工程，然后再使用估计器进行模型训练、预测及评估，它们相互配合，共同完成机器学习任务。

机器学习的管道（Pipeline）机制正是通过将多个转换器和估计器依次连接，构建完整的数据处理和模型训练流程。在 Python 的 scikit - learn 库中，可使用 Pipeline 类来组织和连接不同的转换器与估计器，它提供了简洁的方式来定义和管理机器学习任务流程，方便实现代码复用。

2.优势

防止数据泄露：使用交叉验证时，管道会自动在每个折叠内独立进行预处理（fit/transform），确保测试集数据不会在训练过程中被 “偷看”，从而保证模型评估的准确性。

简化超参数调优：可以方便地同时调整预处理步骤和模型参数。比如将管道与网格搜索结合，可以同时搜索预处理步骤和模型的超参数组合，从而找到整体最优的参数配置。

3.分步代码

无pipeline代码：

import pandas as pd
import pandas as pd   
import numpy as np     
import matplotlib.pyplot as plt   
import seaborn as sns  
import warnings
warnings.filterwarnings("ignore")
plt.rcParams['font.sans-serif'] = ['SimHei'] 
plt.rcParams['axes.unicode_minus'] = False    
data = pd.read_csv('data.csv')   

# 先筛选字符串变量
discrete_features = data.select_dtypes(include=['object']).columns.tolist()
# Home Ownership 标签编码
home_ownership_mapping = {
    'Own Home': 1,
    'Rent': 2,
    'Have Mortgage': 3,
    'Home Mortgage': 4
}
data['Home Ownership'] = data['Home Ownership'].map(home_ownership_mapping)

# Years in current job 标签编码
years_in_job_mapping = {
    '< 1 year': 1,
    '1 year': 2,
    '2 years': 3,
    '3 years': 4,
    '4 years': 5,
    '5 years': 6,
    '6 years': 7,
    '7 years': 8,
    '8 years': 9,
    '9 years': 10,
    '10+ years': 11
}
data['Years in current job'] = data['Years in current job'].map(years_in_job_mapping)

# Purpose 独热编码，记得需要将bool类型转换为数值
data = pd.get_dummies(data, columns=['Purpose'])
data2 = pd.read_csv("data.csv") 
list_final = [] 
for i in data.columns:
    if i not in data2.columns:
       list_final.append(i) 
for i in list_final:
    data[i] = data[i].astype(int) 

# Term 0 - 1 映射
term_mapping = {
    'Short Term': 0,
    'Long Term': 1
}
data['Term'] = data['Term'].map(term_mapping)
data.rename(columns={'Term': 'Long Term'}, inplace=True) 

# 筛选连续特征
continuous_features = data.select_dtypes(include=['int64', 'float64']).columns.tolist()  

# 连续特征用众数补全
for feature in continuous_features:     
    mode_value = data[feature].mode()[0]          
    data[feature].fillna(mode_value, inplace=True)         

# 很多调参函数自带交叉验证，所以这里只划分一次数据集
from sklearn.model_selection import train_test_split
X = data.drop(['Credit Default'], axis=1) 
y = data['Credit Default']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)  

from sklearn.ensemble import RandomForestClassifier 
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score 
from sklearn.metrics import classification_report, confusion_matrix 
import warnings 
warnings.filterwarnings("ignore") 

print("--- 1. 默认参数随机森林 (训练集 -> 测试集) ---")
import time 
start_time = time.time() 
rf_model = RandomForestClassifier(random_state=42)
rf_model.fit(X_train, y_train) 
rf_pred = rf_model.predict(X_test) 
end_time = time.time() 

print(f"训练与预测耗时: {end_time - start_time:.4f} 秒")
print("\n默认随机森林 在测试集上的分类报告：")
print(classification_report(y_test, rf_pred))
print("默认随机森林 在测试集上的混淆矩阵：")
print(confusion_matrix(y_test, rf_pred))

pipeline代码步骤

1.导入库与数据加载

# 导入基础库
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import time 
import warnings

# 忽略警告
warnings.filterwarnings("ignore")

# 设置中文字体和负号正常显示
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 导入 Pipeline 和相关预处理工具
from sklearn.pipeline import Pipeline 
# 用于将不同的预处理应用于不同的列
from sklearn.compose import ColumnTransformer 
# 用于数据预处理（有序编码、独热编码、标准化）
from sklearn.preprocessing import OrdinalEncoder, OneHotEncoder, StandardScaler
# 用于处理缺失值
from sklearn.impute import SimpleImputer 

# 导入机器学习模型和评估工具
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix 
from sklearn.model_selection import train_test_split 

# --- 加载原始数据 ---
data = pd.read_csv('data.csv')
print("原始数据加载完成，形状为:", data.shape)
# 输出：
原始数据加载完成，形状为: (7500, 18)

2.分离特征和标签，划分数据集

y = data['Credit Default'] 
X = data.drop(['Credit Default'], axis=1) 

print("\n特征和标签分离完成。")
print("特征 X 的形状:", X.shape)
print("标签 y 的形状:", y.shape)

# 按照8:2划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) 

print("\n数据集划分完成 (预处理之前)。")
print("X_train 形状:", X_train.shape)
print("X_test 形状:", X_test.shape)
print("y_train 形状:", y_train.shape)
print("y_test 形状:", y_test.shape)

3.定义预处理步骤

# --- 定义不同列的类型和它们对应的预处理步骤 ---
# 这些定义是基于原始数据 X 的列类型来确定的

# 识别原始的 object 列 (对应原代码中的 discrete_features)
object_cols = X.select_dtypes(include=['object']).columns.tolist()
# 识别原始的非 object 列 (通常是数值列)
numeric_cols = X.select_dtypes(exclude=['object']).columns.tolist()

# 有序分类特征 (对应之前的标签编码)
ordinal_features = ['Home Ownership', 'Years in current job', 'Term']
# OrdinalEncoder 会根据类别出现顺序或者预先设定顺序，为每个类别分配从0开始的整数编码。
ordinal_categories = [
    ['Own Home', 'Rent', 'Have Mortgage', 'Home Mortgage'],
    ['< 1 year', '1 year', '2 years', '3 years', '4 years', '5 years', '6 years', '7 years', '8 years', '9 years', '10+ years'], 
    ['Short Term', 'Long Term'] 
]
# 构建处理有序特征的 Pipeline: 先填充缺失值，再进行有序编码
ordinal_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充分类特征的缺失值
    ('encoder', OrdinalEncoder(categories=ordinal_categories, # 进行有序编码handle_unknown='use_encoded_value', unknown_value=-1)) # 遇到未知类别，将其编码为-1
])
print("有序特征处理 Pipeline 定义完成。")


# 标称分类特征 (即无序分类特征，对应之前的独热编码)
nominal_features = ['Purpose'] 
# 构建处理标称特征的 Pipeline: 先填充缺失值，再进行独热编码
nominal_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充分类特征的缺失值
    ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False)) # 进行独热编码, sparse_output=False 使输出为密集数组
])
print("标称特征处理 Pipeline 定义完成。")


# 连续特征 (对应之前的众数填充 + 添加标准化)
# 从所有列中排除掉分类特征，得到连续特征列表
# continuous_features = X.columns.difference(object_cols).tolist()# 原始X中非object的列
# 也可以直接从所有列中排除已知的有序和标称特征
continuous_features = [f for f in X.columns if f not in ordinal_features + nominal_features]

# 构建处理连续特征的 Pipeline: 先填充缺失值，再进行标准化
continuous_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充缺失值
    ('scaler', StandardScaler()) # 标准化
])
print("连续特征处理 Pipeline 定义完成。")

# --- 构建 ColumnTransformer ---
# 将不同的预处理应用于对应的特征列上，构造一个完备的转化器
# ColumnTransformer 接收一个 transformers 列表，每个元素是 (名称, 转换器对象, 列名列表)
preprocessor = ColumnTransformer(
    transformers=[
        ('ordinal', ordinal_transformer, ordinal_features), 
        # 对 ordinal_features 列应用 ordinal_transformer
        ('nominal', nominal_transformer, nominal_features), 
        # 对 nominal_features 列应用 nominal_transformer
        ('continuous', continuous_transformer, continuous_features) 
        # 对 continuous_features 列应用 continuous_transformer
    ],
    remainder='passthrough' # 如何处理没有在上面列表中指定的列。
                            # 'passthrough' 表示保留这些列，不做任何处理。
                            # 'drop' 表示丢弃这些列，适用于某些列不需要参与建模的情况
)
print("\nColumnTransformer (预处理器) 定义完成。")
# print(preprocessor) # 可以打印 preprocessor 对象看看它的结构

4.构建完整pipeline

# --- 构建完整的 Pipeline ---
# 将预处理器和模型串联起来
# 使用原代码中 RandomForestClassifier 的默认参数和 random_state
pipeline = Pipeline(steps=[
    ('preprocessor', preprocessor), 
# 第一步：应用所有的预处理 (我们刚刚定义的 ColumnTransformer 对象)
    ('classifier', RandomForestClassifier(random_state=42)) 
# 第二步：随机森林分类器 (使用默认参数和指定的 random_state)
])
print("\n完整的 Pipeline 定义完成。")
# print(pipeline) # 可以打印 pipeline 对象看看它的结构

5.使用pipeline进行训练与评估

# --- 使用 Pipeline 在划分好的训练集和测试集上评估 ---
print("\n--- 1. 默认参数随机森林 (训练集 -> 测试集) ---") 

start_time = time.time() 
# 在原始的 X_train, y_train 上拟合整个Pipeline
# Pipeline会自动按顺序执行 preprocessor 的 fit_transform(X_train)，
# 然后用处理后的数据和 y_train 拟合 classifier
pipeline.fit(X_train, y_train)

# 在原始的 X_test 上进行预测
# Pipeline会自动按顺序执行 preprocessor 的 transform(X_test)，
# 然后用处理后的数据进行 classifier 的 predict
pipeline_pred = pipeline.predict(X_test)
end_time = time.time() 

print(f"训练与预测耗时: {end_time - start_time:.4f} 秒") 
print("\n默认随机森林 在测试集上的分类报告：") 
print(classification_report(y_test, pipeline_pred))
print("默认随机森林 在测试集上的混淆矩阵：") 
print(confusion_matrix(y_test, pipeline_pred))

4.纯享版代码

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import time 
import warnings
warnings.filterwarnings("ignore")
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False 

# 导入 Pipeline 和相关预处理工具
from sklearn.pipeline import Pipeline 
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OrdinalEncoder, OneHotEncoder, StandardScaler 
from sklearn.impute import SimpleImputer 

# 导入机器学习相关库
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix, accuracy_score, precision_score, recall_score, f1_score
from sklearn.model_selection import train_test_split 

# 加载原始数据
data = pd.read_csv('data.csv')

# 分离特征和标签
y = data['Credit Default']
X = data.drop(['Credit Default'], axis=1)

# 划分训练集和测试集 (在任何预处理之前划分) 
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 定义不同列的类型和对应的预处理步骤 (这些将被放入 Pipeline 的 ColumnTransformer 中)
object_cols = X.select_dtypes(include=['object']).columns.tolist()
numeric_cols = X.select_dtypes(exclude=['object']).columns.tolist()
# 有序分类特征（标签编码）
ordinal_features = ['Home Ownership', 'Years in current job', 'Term']
ordinal_categories = [
    ['Own Home', 'Rent', 'Have Mortgage', 'Home Mortgage'], 
    ['< 1 year', '1 year', '2 years', '3 years', '4 years', '5 years', '6 years', '7 years', '8 years', '9 years', '10+ years'], 
    ['Short Term', 'Long Term'] 
]
ordinal_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充分类特征的缺失值
    ('encoder', OrdinalEncoder(categories=ordinal_categories, handle_unknown='use_encoded_value', unknown_value=-1))
])
# 标称分类特征（独热编码）
nominal_features = ['Purpose'] 
nominal_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充分类特征的缺失值
    ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False)) # sparse_output=False 使输出为密集数组
])
# 连续特征（标准化）
continuous_features = X.columns.difference(object_cols).tolist() 
continuous_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')), 
    ('scaler', StandardScaler()) 
])

# 构建 ColumnTransformer 将不同的预处理应用于不同的列子集，构造一个完备的转化器
preprocessor = ColumnTransformer(
    transformers=[
        ('ordinal', ordinal_transformer, ordinal_features),
        ('nominal', nominal_transformer, nominal_features),
        ('continuous', continuous_transformer, continuous_features)
    ],
    remainder='passthrough' 
)

# 构建完整的 Pipeline，将预处理器和模型串联起来
pipeline = Pipeline(steps=[
    ('preprocessor', preprocessor), 
    ('classifier', RandomForestClassifier(random_state=42)) 
])

# 使用 Pipeline 在划分好的训练集和测试集上评估
print("默认参数随机森林 (训练集 -> 测试集) ") 
start_time = time.time() 
pipeline.fit(X_train, y_train)
pipeline_pred = pipeline.predict(X_test)
end_time = time.time() 

print(f"训练与预测耗时: {end_time - start_time:.4f} 秒")
print("\n默认随机森林 在测试集上的分类报告：")
print(classification_report(y_test, pipeline_pred))
print("默认随机森林 在测试集上的混淆矩阵：") 
print(confusion_matrix(y_test, pipeline_pred))

5.pipeline的嵌套

在机器学习中，管道（Pipeline）可以进行一定程度的嵌套连接，这种方式能让数据处理和模型构建流程更加灵活和复杂。

直接嵌套 Pipeline 对象：可以将一个 Pipeline 作为另一个 Pipeline 的步骤。比如先定义一个用于数据预处理的 Pipeline，然后将其作为新 Pipeline 的一部分，与模型训练步骤连接起来。

使用 ColumnTransformer 与 Pipeline 嵌套：ColumnTransformer 本身可以看作是一种特殊的 “管道”，它能对不同列的数据应用不同的预处理操作。ColumnTransformer 可以与 Pipeline 配合使用，实现更复杂的数据处理流程。例如，对不同类型的特征列（数值列、分类列等）分别应用不同的 Pipeline 进行预处理，然后再将这些处理结果合并起来，输入到模型中。

6.函数与类的区别（pipeline）

在机器学习的 pipeline 场景中，虽然很多操作函数与类都能实现，但类具有独特优势。如pipeline 中的转换器需遵循特定接口规范，通常要实现 fit 和 transform 方法，类可自然定义这两个方法，函数则不易在结构上适配这种规范。

7.pipeline 的优势

参数处理灵活：使用 pipeline 时能在外部统一设置各步骤参数，即便某个步骤在特定情况下无需特定参数，pipeline 也能忽略该无用参数设置，不会因特殊参数设置导致整个流程崩溃。

操作与参数分离：其核心优势之一是将操作（如数据标准化、模型训练）和操作依赖的参数（如标准化的均值、标准差，模型学习率）分离开，开发者可专注于在 pipeline 外部参数列表设置参数来优化模型，无需耗费大量精力在代码结构和操作实现细节上。

契合复杂项目管理：复杂 Python 项目需拆分代码到多个文件以提高模块化和可管理性，pipeline 将操作和参数分离的思想与之契合。每个 pipeline 步骤可对应单独 Python 文件（模块），主程序通过 pipeline 连接这些模块并在外部统一设置参数，项目规模扩大需修改操作或调整参数时，定位和处理更清晰，不会因代码分散而混乱。

@浙大疏锦行