一、Pipeline基本知识

Pipeline 对象将不同计算步骤串联起来执行，方便、快速、统一、保险！

构造一个Pipeline对象时，需要输入一个由多个二元tuple构成的list，每个tuple的第一个元素可以是任意自定义字符串，第二元素表示一个已经存在的计算步骤，例：
my_steps = [(name1, algo1), (name2, algo2), (name3, algo3)...] # 定义计算步骤list
pipe = Pipline(step=my_steps) # 生成Pipline对象

这些计算步骤(algo)通常有3种类型：变换器（transformer）、估计器（estimator）和'passthrough'

• 变换器（transformer）：
  （1）变换器通常具有fit()、transform()、fit_transform()三个方法，StandardScaler()、PCA()等都属于变换器。
  （2）其中fit()根据输入数据X计算参数S（并保存S），然后 transform()根据参数S对数据（X或X_new）做变换，并返回变换结果。
  （3）Pipeline对象中，多个变换器串联，其本质是将每个变换器按照步骤（2）的方式串起来，即：
     第1个变换器T1：利用T1.fit(X)计算并保存S1，然后基于S1计算 result1 = T1.transform(X)；
     第2个变换器T2：利用T2.fit(result1)计算并保存S2，然后基于S2计算 result2 = T2.transform(result1)；
     ……
  （4）Pipeline对象中需要保证前一个变换器的输出可以作为后一个变换器的输入，否则报错。

• 估计器（estimator）：
  （1）通常是机器学习算法（比如线性回归、支持向量机、决策树等），通常有fit()和predict()方法。
  （2）在一个Pipeline中估计器通常只有一个，且放在最后。

• 'passthrough'：
  表示这一步啥也不干！

Pipeline的使用方法：

• Pipeline对象（本文用pipe表示）可以像普通变换器一样使用pipe.fit()、pipe.transform()、pipe.fit_transform()或像普通估计器一样使用pipe.predict()方法，其前提是pipe内部这些变换器都实现了前三种方法和估计器实现了第四种方法，否则出错。
• Pipeline的使用模式通常有2种：
  （1）模式1：只有n个变换器（transformer）没有估计器（estimator）
     这种模式主要用于对数据进行变换，但不包括对变换后的结果进行分类或拟合等操作（这是估计器干的事）。
  模式1.png
  （2）模式2：前n个变换器（transformer）+ 最后一个估计器（estimator）
     这种模式就是在第（1）种模式上加了估计器，从而实现分类或拟合（估计器一般只有1个，且放在最后，否则会引起错误。）
  模式2.png
  （注意：上述只是伪代码，实际使用时还有其他各种参数，同时也还有很多其他方法！）
  （3）Pipline的高级使用方法：请搜索其他文档！
     与GridSearchCV、make_union、FeatureUnion结合。

二、Pipeline内部计算过程窥探（包括一些结论）

下面我们自定义两个变换器（实际上也是估计器，因为我们都定义了predict()函数），并组成Pipeline，来窥探其内部执行过程！

import numpy as np
from sklearn.base import TransformerMixin, BaseEstimator
from sklearn.pipeline import Pipeline

定义变换器：

说明：我们用 self.flag 变量来代表‘计算参数S’，即fit(X)根据输入X计算得到的。

# 自定义变换器1
class T1(BaseEstimator, TransformerMixin):
    def __init__(self):
        self.flag = 'T1原始参数S1_old！'

    def fit(self, X, y=None):
        self.flag = 'T1 fit() 计算的新参数S1_new！'
        print(f'Run T1 fit(), 输入X={X}，计算并保存新参数S1_new！')
        return self

    def transform(self, X, y=None):
        print(f'Run T1 transform(), 输入X={X}, 用 {self.flag}')
        return X*2
    
    def predict(self, X):
         print(f'Run T1 predict(), 输入X={X}, 用 {self.flag}')

# 自定义变换器2
class T2(BaseEstimator, TransformerMixin):
    
    def __init__(self):
        self.flag = '原始参数S2_old！'

    def fit(self, X, y=None):
        self.flag = 'T2 fit() 计算的新参数S2_new！'
        print(f'Run T2 fit(), 输入X={X}，计算并保存新参数S2_new！')
        return self

    def transform(self, X, y=None):
        print(f'Run T2 transform(), 输入X={X}, 用 {self.flag}')
        return X*4
    
    def predict(self, X):
         print(f'Run T2 predict(), 输入X={X}, 用 {self.flag}')

例1：pipe.fit()→pipe.transform()→pipe.predict()

X = np.array([1,2,3,4,5])  # 数据准备

pipe = Pipeline([('name1', T1()), ('name2', T2())]) # 生成Pipline对象

pipe.fit(X)
print('#######################################分割线1################################################')
result = pipe.transform(X)
print(result)
print('#######################################分割线2################################################')
pipe.predict(X)

输出：
Run T1 fit(), 输入X=[1 2 3 4 5]，计算并保存新参数S1_new！
Run T1 transform(), 输入X=[1 2 3 4 5], 用 T1 fit() 计算的新参数S1_new！
Run T2 fit(), 输入X=[ 2  4  6  8 10]，计算并保存新参数S2_new！
#######################################分割线1################################################
Run T1 transform(), 输入X=[1 2 3 4 5], 用 T1 fit() 计算的新参数S1_new！
Run T2 transform(), 输入X=[ 2  4  6  8 10], 用 T2 fit() 计算的新参数S2_new！
[ 8 16 24 32 40]
#######################################分割线2################################################
Run T1 transform(), 输入X=[1 2 3 4 5], 用 T1 fit() 计算的新参数S1_new！
Run T2 predict(), 输入X=[ 2  4  6  8 10], 用 T2 fit() 计算的新参数S2_new！

• 结论1：执行pipe.fit() 方法，实际上执行了所有内部变换器的fit()和transform()，但不会执行最后一个变换器的 transform() 方法
  这一点从分割线1前面的输出结果就可以看出！
  这是合理的，因为pipe.fit()阶段的主要目的是根据输入数据X计算所有步骤的参数S！这个过程在前面变换器(transformer)-(3)节已有详细描述。
  根据这个逻辑，最后一个变换器只要利用前一个变换器的transform()输出，再通过自己的fit()计算自己的参数S就可以了，它的transform()没有执行的必要了，因为后面没有其他步骤需要计算参数了！
• 结论2：使用pipe.fit() 方法后，会自动保存计算参数，留作后用
  从所有输出结果可知，并没有出现原始参数S1_old！或原始参数S2_old！，都是S1_new或S2_new，说明在执行pipe.fit(X)时，确实保留了新计算参数！
• 结论3：在pipe.fit() 后调用pipe.transform()，则会把所有计算步骤的transform()都执行一遍，并输出最终结果，且每一步用的是每个计算步骤的fit()所计算和保存的参数S
  这点从分割线1和2之间的输出结果可以看出！
• 结论4：执行pipe.predict()时，实际上执行的是pipe中最后一个计算步骤的predict()（也就是说如果要用pipe.predict()，那么就要求最后一个计算步骤必须有predict()这个方法，否则报错）；且在执行predict()之前，排在其前面的所有步骤的transform()方法都会被执行一遍
  这个结论的前半部分可以通过将T2定义中的 def predict(self, X):部分全部删除来验证，结果会报错；结论的后半部分，从分割线2下的输出可以看出！很显然，T2 的 transform()没有被执行，因为它与最后一个predict()处于同一个类中，pipeline首先将其视为估计器，自动忽略了T2 的 transform()。实际中，真实的估计器，都是SVM、LR等机器学习方法，通常不会出现transform()和predict()同时出现在一个类中的情况。

例2：pipe.fit_transform()→pipe.predict()

X = np.array([1,2,3,4,5])  # 数据准备

pipe = Pipeline([('name1', T1()), ('name2', T2())]) # 生成Pipline对象

result = pipe.fit_transform(X)
print(result)
print('#######################################分割线1################################################')
pipe.predict(X)

输出：
Run T1 fit(), 输入X=[1 2 3 4 5]，计算并保存新参数S1_new！
Run T1 transform(), 输入X=[1 2 3 4 5], 用 T1 fit() 计算的新参数S1_new！
Run T2 fit(), 输入X=[ 2  4  6  8 10]，计算并保存新参数S2_new！
Run T2 transform(), 输入X=[ 2  4  6  8 10], 用 T2 fit() 计算的新参数S2_new！
[ 8 16 24 32 40]
#######################################分割线1################################################
Run T1 transform(), 输入X=[1 2 3 4 5], 用 T1 fit() 计算的新参数S1_new！
Run T2 predict(), 输入X=[ 2  4  6  8 10], 用 T2 fit() 计算的新参数S2_new！

• 结论5：pipe.fit_transform(X)除了执行了pipe.fit(X)和pipe.transform(X)所执行的所有步骤外，还执行了最后一个变换器的transform()，并输出了最终结果
  比较例1和2两种情况分割线1前面部分就知道了。
• 结论6：pipe.fit()、pipe.transform()、pipe.fit_transform()这几个方法要正常执行，要求pipe中每个计算步骤内部都实现了fit()和transform()方法，否则会报错。
  这点可以通过删除T1或T2中的一个或多个fit()或transform()函数得到验证。

• 一个pipeline对象可以作为另一个pipeline对象的一个计算步骤。
  例子省略。

三、一个比较完整的Pipeline例子

对鸢尾花数据集进行分类：

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

from sklearn.datasets import load_iris  # 自带的样本数据集
from sklearn.model_selection import train_test_split

iris = load_iris()  # 导入数据

X = iris.data  # 150个样本，4个属性
y = iris.target # 150个类标号

# 将数据集划分为训练集和验证集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 生成pipeline对象
pipe = Pipeline([('preprocessing','passthrough'),
                 ('sc', StandardScaler()),
                 ('pca', PCA(n_components=2)),
                 ('clf', LogisticRegression(random_state=1))])

# 利用pipeline对象训练模型
pipe.fit(X_train, y_train)

# 利用测试数据对模型进行评价
pipe.score(X_test,y_test)
输出：
0.9

# 利用训练好的模型对新数据进行预测（这里用测试数据代替）
pipe.predict(X_test)
输出：
array([1, 0, 2, 1, 2, 0, 1, 2, 2, 1, 2, 0, 0, 0, 0, 1, 2, 1, 1, 2, 0, 1, 0, 2, 2, 2, 2, 2, 0, 0])

上述例子中，在定义pipe对象时，用了四个计算步骤：其中第1个是'passthrough'，啥也不做；第2、3个分别是标准化处理和PCA分解，都属于变换器（transformer）；第4个是逻辑回归，属于估计器（estimator）。

我们继续做实验，在pipe步骤中做两种修改：1）估计器不放在最后位置；2）有两个估计器，如下所示：

# 修改1：把估计器（逻辑回归）放在非最后位置
pipe = Pipeline([('preprocessing','passthrough'),
                 ('sc', StandardScaler()),
                 ('clf', LogisticRegression(random_state=1)),
                 ('pca', PCA(n_components=2))])

# 修改2：把增加一个估计器（逻辑回归clf和clf2）
pipe = Pipeline([('preprocessing','passthrough'),
                 ('sc', StandardScaler()),
                 ('pca', PCA(n_components=2)),
                 ('clf', LogisticRegression(random_state=1)),
                 ('clf2', LogisticRegression(random_state=1))])

上述两种改变，在定义时都不会报错，但在执行pipe.fit(X_train, y_train)时都会报错：

pipe.fit(X_train, y_train)
输出：
 TypeError: All intermediate steps should be transformers and implement fit and transform or be the string 'passthrough' 'LogisticRegression(random_state=1)' (type <class 'sklearn.linear_model._logistic.LogisticRegression'>) doesn't

由此可见：

• 估计器（estimator）通常需要放在Pipeline对象的最后位置；
• 一个Pipeline对象中不能有多个估计器（estimator）。

参考：
https://blog.csdn.net/nkufang/article/details/129973061