sklearn中Pipeline包的用法
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
"""
使用方法:
输入一连串数据挖掘步骤,最后一步必须是估计器,前几步是转换器
输入的数据集经过转换器的处理后,输出的结果作为下一步的输入
最后用于估计器进行分类
每一步都是元祖(‘名称’，步骤)来表示
流水线功能:
跟踪记录各步骤操作
对各步骤进行封装
确保代码的复杂程度不至于超出掌控范围
"""
pip = Pipeline([('sc', StandardScaler()),
                ('pca', PCA(n_components=2)),
                ('clf', LogisticRegression(random_state=666))
                ])
pip.fit(x_train, y_train)
print('Test accuracy is %.3f'% pip.score(x_test, y_test))

本文和大家分享的主要是机器学习中 sklearn中pipeline相关内容，一起来看看吧，希望对大家学习机器学习有所帮助。
　　如下图所示，利用pipeline我们可以方便的减少代码量同时让机器学习的流程变得直观，
例如我们需要做如下操作，容易看出，训练测试集重复了代码，
　　vect = CountVectorizer()tfidf = TfidfTransformer()clf = SGDClassifier()
　　vX = vect.fit_transform(Xtrain)tfidfX = tfidf.fit_transform(vX)predicted = clf.fit_predict(tfidfX)
　　# Now evaluate all steps on test setvX = vect.fit_transform(Xtest)tfidfX = tfidf.fit_transform(vX)predicted = clf.fit_predict(tfidfX)
　　利用pipeline，上面代码可以抽象为，
　　pipeline = Pipeline([
　　('vect', CountVectorizer()),
　　('tfidf', TfidfTransformer()),
　　('clf', SGDClassifier()),
　　])
　　predicted = pipeline.fit(Xtrain).predict(Xtrain)
　　# Now evaluate all steps on test set
　　predicted = pipeline.predict(Xtest)
　　注意，pipeline最后一步如果有predict()方法我们才可以对pipeline使用fit_predict()，同理，最后一步如果有transform()方法我们才可以对pipeline使用fit_transform()方法。
　　使用pipeline做cross validation
　　看如下案例，即先对输入手写数字的数据进行PCA降维，再通过逻辑回归预测标签。其中我们通过pipeline对
　　PCA的降维维数n_components和逻辑回归的正则项C大小做交叉验证，主要步骤有：
　　1. 依次实例化各成分对象如 pca = decomposition.PCA()
　　2. 以(name, object)的tuble为元素组装pipeline如 Pipeline(steps=[('pca', pca), ('logistic', logistic)])
　　3. 初始化CV参数如 n_components = [20, 40, 64]
　　4. 实例化CV对象如 estimator = GridSearchCV(pipe, dict(pca__n_components=n_components, logistic__C=Cs)) ，其中注意参数的传递方式，即key为pipeline元素名+函数参数
　　import numpy as npimport matplotlib.pyplot as pltfrom sklearn import linear_model, decomposition, datasetsfrom sklearn.pipeline import Pipelinefrom sklearn.model_selection import GridSearchCV
　　logistic = linear_model.LogisticRegression()
　　pca = decomposition.PCA()
　　pipe = Pipeline(steps=[('pca', pca), ('logistic', logistic)])
　　digits = datasets.load_digits()
　　X_digits = digits.data
　　y_digits = digits.target
　　# Prediction
　　n_components = [20, 40, 64]
　　Cs = np.logspace(-4, 4, 3)
　　pca.fit(X_digits)
　　estimator = GridSearchCV(pipe,
　　dict(pca__n_components=n_components, logistic__C=Cs))
　　estimator.fit(X_digits, y_digits)
　　plt.figure(1, figsize=(4, 3))
　　plt.clf()
　　plt.axes([.2, .2, .7, .7])
　　plt.plot(pca.explained_variance_, linewidth=2)
　　plt.axis('tight')
　　plt.xlabel('n_components')
　　plt.ylabel('explained_variance_')
　　plt.axvline(
　　estimator.best_estimator_.named_steps['pca'].n_components,
　　linestyle=':',
　　label='n_components chosen')
　　plt.legend(prop=dict(size=12))
　　plt.show()
 
来源：网络

转载请注明出处：https://leytton.blog.csdn.net/article/details/101351814

如果本文对您有所帮助，请点个赞让我知道哦 😃
在本教程中，你将学习如何使用pipeline来清理你的建模代码。
1、介绍
Pipeline是一种简单的方法，能让你的数据预处理和建模步骤一步到位。
很多数据科学家没有使用pipeline来建模，但pipeline有很多重要好处。包含：

更精简的代码：考虑到数据处理时会造成混乱，使用pipeline不需要在每个步骤都特别注意训练和验证数据。
更少的Bug：错误应用和忘记处理步骤的概率更小。
更易产品化：把模型转化成规模化发布原型是比较难的，再此我们不做过多讨论，但pipeline对这个有帮助。
模型验证更加多样化：你将会在下一个课程中看到交叉验证的案例。

2、案例
跟前面教程一样，我们将会使用 Melbourne Housing 数据集
我们不会关注数据加载步骤。假设你已经拥有了X_train、X_valid、y_train和y_valid中的训练和验证数据。
我们先使用head()方法瞄一眼训练数据。注意这些数据保护分类数据和缺失数据。使用pipelines将会很方便处理这两者。
X_train.head()

输出结果：
	Type 	Method 	Regionname 	Rooms 	Distance 	Postcode 	Bedroom2 	Bathroom 	Car 	Landsize 	BuildingArea 	YearBuilt 	Lattitude 	Longtitude 	Propertycount
12167 	u 	S 	Southern Metropolitan 	1 	5.0 	3182.0 	1.0 	1.0 	1.0 	0.0 	NaN 	1940.0 	-37.85984 	144.9867 	13240.0
6524 	h 	SA 	Western Metropolitan 	2 	8.0 	3016.0 	2.0 	2.0 	1.0 	193.0 	NaN 	NaN 	-37.85800 	144.9005 	6380.0
8413 	h 	S 	Western Metropolitan 	3 	12.6 	3020.0 	3.0 	1.0 	1.0 	555.0 	NaN 	NaN 	-37.79880 	144.8220 	3755.0
2919 	u 	SP 	Northern Metropolitan 	3 	13.0 	3046.0 	3.0 	1.0 	1.0 	265.0 	NaN 	1995.0 	-37.70830 	144.9158 	8870.0
6043 	h 	S 	Western Metropolitan 	3 	13.3 	3020.0 	3.0 	1.0 	2.0 	673.0 	673.0 	1970.0 	-37.76230 	144.8272 	4217.0

我们通过三个步骤来使用pipelines：
步骤1：定义处理步骤
与pipeline将预处理与建模步骤打包一样，我们使用ColumnTransformer类来将不同的步骤打包在一起。下面的代码做了两件事情：

填充缺失数据为数值类型（用均值等方法填充数值类型数据）
用one-hot编码来填充分类缺失数据

from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import OneHotEncoder

# 预处理数值类型数据
numerical_transformer = SimpleImputer(strategy='constant')

# 预处理分类数据
categorical_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('onehot', OneHotEncoder(handle_unknown='ignore'))
])

# 将处理步骤打包
preprocessor = ColumnTransformer(
    transformers=[
        ('num', numerical_transformer, numerical_cols),
        ('cat', categorical_transformer, categorical_cols)
    ])

步骤2：定义模型
接下来我们使用熟悉的RandomForestRegressor类定义一个随机森林模型。
from sklearn.ensemble import RandomForestRegressor

model = RandomForestRegressor(n_estimators=100, random_state=0)

步骤3：创建和评估Pipeline
最后，我们使用Pipeline类来定义一个打包预处理和建模步骤的pipeline。这里有些重点需要注意：

使用pipeline，我们预处理训练数据以及拟合模型只有了一行代码。（相反，如果不使用pipeline，我们需要做填充、编码、和模型训练的步骤。如果我们需要同时处理数值和分类变量，这将非常繁琐！）
我们在调用predict()指令时使用的X_valid中包含未经处理的特征值，pipeline会在预测前自动进行预处理。（然而，如果不使用pipeline，在预测前，我们需要记住对验证数据进行预处理）。

from sklearn.metrics import mean_absolute_error

# 在pipeline中打包预处理和建模代码
my_pipeline = Pipeline(steps=[('preprocessor', preprocessor),
                              ('model', model)
                             ])

# 预处理训练数据，拟合模型 
my_pipeline.fit(X_train, y_train)

# 预处理验证数据, 获取预测值
preds = my_pipeline.predict(X_valid)

# 评估模型
score = mean_absolute_error(y_valid, preds)
print('MAE:', score)

输出结果：
MAE: 160679.18917034855

3、结论
Pipeline在机器学习代码清理和规避错误中，尤其是复杂数据预处理的工作流，非常实用。
4、去吧，皮卡丘
在接下来的练习中，使用pipeline来体验高级数据预处理技术并改善你的预测！
原文：

https://www.kaggle.com/alexisbcook/pipelines

Kubeflow项目使得在Kubernetes上部署机器学习的工作流更加容易，可移植并且可扩展。其中Pipeline是Kubeflow的一个核心模块，可以定义机器学习各个步骤的工作流，按照编排来进行执行。

我在学习Kubeflow的过程中，发现网上关于Pipeline方面的介绍比较少，官网的例子则比较晦涩，而且大都是要结合GCP环境来部署的。对于我们个人学习来说，没有一个比较简明的教程。我研究了一下，实现了在个人电脑上用kubeadm部署了一个k8s集群，运行Kubeflow，并以预测泰坦尼克号生还者这一个经典的机器学习任务为例，实现了一个pipeline来完成从数据特征转换，模型训练到数据预测的全过程。

数据准备

首先需要准备Titantic的数据集，在网上下载之后，我把数据集保存到了本地目录。要使pipeline能访问数据，我是通过创建一个Mount到本地目录的PV来实现的，之后就可以在pipeline里面通过给component attach pvolume的方式来访问数据了。以下是创建PV的yaml:

apiVersion: v1
kind: PersistentVolume
metadata:
  name: my-pv
spec:
  capacity:
    storage: 1Gi
  volumeMode: Filesystem
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Delete
  storageClassName: local-storage
  local:
    path: /home/abc/data
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - abc-desktop

数据特征转换

这里略去对数据集的探索的过程。根据数据探索的结果，我们需要填补Age, Embarked这2个特征的缺失值，删去缺失值过多的Cabin特征，从Name中提取生成新的特征Title，以及把Sex, Embarked, Title这3个特征转为类别值。具体的生成pipeline component的Python代码如下，程序名称为titantic_preprocess_component.py，运行后即可生成一个yaml文件：

import kfp.components as comp

def titantic_preprocess(
    train_dataset: comp.InputTextFile(str),
    test_dataset: comp.InputTextFile(str),
    processed_train_dataset: comp.OutputBinaryFile(str),
    processed_test_dataset: comp.OutputBinaryFile(str)):

    import pandas as pd
    import re
    from sklearn import preprocessing
    import pickle

    # Based on the EDA reulst to fill the missing age by title
    age_title = {
        'Mr': 32.368090452261306,
        'Mrs': 35.898148148148145,
        'Miss': 21.773972602739725,
        'Master': 4.574166666666667,
        'Dr': 42.0,
        'Ms': 21.773972602739725
    }

    df_train = pd.read_csv(train_dataset, header=0)
    df_test = pd.read_csv(test_dataset, header=0)

    # Get the title from the Name feature and generate a new Title feature
    title_regex = re.compile(r'.*, ([^\.]*).*')
    def getTitle(x):
        result = title_regex.search(x)
        if result:
            return result.group(1)
        else:
            return ''
    df_train['Title'] = df_train['Name'].map(getTitle)
    df_test['Title'] = df_test['Name'].map(getTitle)

    # Fill the null value of age
    for t in age_title.keys():
        df_train.loc[df_train[(df_train['Title']==t)&(df_train['Age'].isnull())].index, 'Age'] = age_title[t]
        df_test.loc[df_test[(df_test['Title']==t)&(df_test['Age'].isnull())].index, 'Age'] = age_title[t]

    # Drop the cabin feature
    df_train = df_train.drop(['Cabin'], axis=1)
    df_test = df_test.drop(['Cabin'], axis=1)

    # Two record of Embarked feature missing, fill with the most frequent value
    df_train.loc[df_train[df_train['Embarked'].isnull()].index, 'Embarked'] = 'S'

    # Drop the PassengerId, Name, Ticket features, as no use for model training
    df_train = df_train.drop(['PassengerId', 'Name', 'Ticket'], axis=1)
    df_test = df_test.drop(['PassengerId', 'Name', 'Ticket'], axis=1)

    # Conver the Sex, Embarked, Title feature to category type
    le_sex = preprocessing.LabelEncoder()
    le_sex.fit(df_train['Sex'])
    df_train['Sex_cat'] = le_sex.transform(df_train['Sex'])
    df_test['Sex_cat'] = le_sex.transform(df_test['Sex'])
    le_embarked = preprocessing.LabelEncoder()
    le_embarked.fit(df_train['Embarked'])
    df_train['Embarked_cat'] = le_embarked.transform(df_train['Embarked'])
    df_test['Embarked_cat'] = le_embarked.transform(df_test['Embarked'])
    le_title = preprocessing.LabelEncoder()
    le_title.fit(df_train['Title'])
    df_train['Title_cat'] = le_title.transform(df_train['Title'])
    df_test.loc[df_test[df_test['Title']=='Ms'].index, 'Title'] = 'Miss'
    df_test.loc[df_test[df_test['Title']=='Dona'].index, 'Title'] = 'Mrs'
    df_test['Title_cat'] = le_title.transform(df_test['Title'])

    # Drop the Sex, Embarked, Title features
    df_train = df_train.drop(['Sex', 'Embarked', 'Title'], axis=1)
    df_test = df_test.drop(['Sex', 'Embarked', 'Title'], axis=1)

    # Output the processed train and test dataset
    pickle.dump(df_train, processed_train_dataset)
    pickle.dump(df_test, processed_test_dataset)

comp.create_component_from_func(
    titantic_preprocess, 
    base_image='gzroy/ml_baseimage',
    output_component_file='titantic_preprocess_component.yaml')

模型训练

特征转换完成后，就可以建立模型来进行训练了。这里我用随机森林来进行训练。程序名称为titantic_train_component.py，运行后同样生成一个yaml文件：

import kfp.components as comp

def titantic_train(
    dataset: comp.InputBinaryFile(str),
    model: comp.OutputBinaryFile(str),
    predict_result: comp.OutputTextFile(str)):

    import pandas as pd
    import pickle
    from sklearn.model_selection import train_test_split
    from sklearn.ensemble import RandomForestClassifier
    import numpy as np

    # Load the preprocessed data
    df_train = pickle.load(dataset)
    
    # Random split the train and test data
    X_train, X_test, y_train, y_test = train_test_split(
        df_train[['Pclass', 'Sex_cat', 'Age', 'SibSp', 'Parch', 'Embarked_cat', 'Title_cat']], 
        df_train['Survived'], 
        test_size=0.2, 
        random_state=123
    )

    # Use the random forest classifer to train the model
    rfc = RandomForestClassifier()
    rfc.fit(X_train, y_train)

    # Output the prediction on test dataset
    y_pred = rfc.predict(X_test)
    test_result = np.concatenate([y_test.to_numpy().reshape([-1,1]), y_pred.reshape([-1,1])], axis=-1)
    test_result_df = pd.DataFrame(test_result, columns=['Label', 'Prediction'])
    test_result_df.to_csv(predict_result, header=True, index=False)

    # Output the model
    pickle.dump(rfc, model)

comp.create_component_from_func(
    titantic_train, 
    base_image='gzroy/ml_baseimage',
    output_component_file='titantic_train_component.yaml')

评估性能

模型训练完成之后，我们需要评估模型的性能，因此这里准备了一个component，根据模型训练之后对验证集的预测值进行准确率的评分：

import kfp.components as comp
from typing import NamedTuple

def produce_metrics(
    predict_file: comp.InputTextFile(str),
    mlpipeline_metrics_path: comp.OutputPath('Metrics')
):
    import json
    import pandas as pd
    from sklearn.metrics import accuracy_score

    df = pd.read_csv(predict_file, header=0)
    accuracy = accuracy_score(df['Label'], df['Prediction'])
    metrics = {
        'metrics': [{
            'name': 'accuracy-score', # The name of the metric. Visualized as the column name in the runs table.
            'numberValue':  accuracy, # The value of the metric. Must be a numeric value.
            'format': "PERCENTAGE",   # The optional format of the metric. Supported values are "RAW" (displayed in raw format) and "PERCENTAGE" (displayed in percentage format).
        }]
    }

    with open(mlpipeline_metrics_path, 'w') as f:
        json.dump(metrics, f)

comp.create_component_from_func(
    produce_metrics, 
    base_image='gzroy/ml_baseimage',
    output_component_file='metrics_component.yaml')

测试集数据预测

模型训练后就可以对测试集的数据进行预测了，代码如下：

import kfp.components as comp

def titantic_predict(
    dataset: comp.InputBinaryFile(str),
    model: comp.InputBinaryFile(str),
    predict_result: comp.OutputTextFile(str)):

    import pandas as pd
    import pickle

    # Load the preprocessed data
    df_test = pickle.load(dataset)
    # Load the model
    rfc = pickle.load(model)
    # Predict
    predict = rfc.predict(df_test[['Pclass', 'Sex_cat', 'Age', 'SibSp', 'Parch', 'Embarked_cat', 'Title_cat']])
    df_test['Prediction'] = predict
    df_test.to_csv(predict_result, header=True, index=False)

comp.create_component_from_func(
    titantic_predict, 
    base_image='gzroy/ml_baseimage',
    output_component_file='titantic_predict_component.yaml')

生成模块的基础镜像

在以上模块中都需要基于基础镜像来运行，以下是构造镜像的Dockerfile：

FROM ubuntu:18.04
RUN apt update \
  && apt install python3.8 -y \
  && apt install python-pip3 -y \
  && rm /usr/bin/python3 \
  && ln -s /usr/bin/python3.8 /usr/bin/python3 \
  && pip3 install pandas -y \
  && pip3 install sklearn -y \


 

生成Pipeline

现在可以定义一个Pipeline，加载之前创建的模块的yaml文件，按照模块的运行顺序进行编排。这里定义了一个produce_data_op，用于生成一个简单的模块，访问pipeline volume里面的数据集，并传给下一个模块：

import kfp
from kfp import dsl

@kfp.dsl.pipeline(
    name='Titantic training pipeline',
    description='My machine learning pipeline'
)
def titantic_pipeline():
    vop = dsl.VolumeOp(
        name="volume_creation",
        resource_name="titantic_pvc",
        storage_class="local-storage",
        modes=["ReadWriteOnce"],
        size="1Gi",
        volume_name="my-pv"
    )

    def produce_data_op(volume):
        return dsl.ContainerOp(
            name="Titantic-Data",
            image="ubuntu:18.04",
            file_outputs={
                'train_dataset': '/data/titantic/train.csv',
                'test_dataset': '/data/titantic/test.csv'
            },
            pvolumes={"/data": volume}
    )

    produce_data_task = produce_data_op(vop.volume)

    preprocess_op = kfp.components.load_component_from_file('titantic_preprocess_component.yaml')
    preprocess_task = preprocess_op(
        produce_data_task.outputs['train_dataset'], 
        produce_data_task.outputs['test_dataset'])

    train_op = kfp.components.load_component_from_file('titantic_train_component.yaml')
    train_task = train_op(preprocess_task.outputs['processed_train_dataset'])

    metrics_op = kfp.components.load_component_from_file('metrics_component.yaml')
    metrics_task = metrics_op(train_task.outputs['predict_result'])

    predict_op = kfp.components.load_component_from_file('titantic_predict_component.yaml')
    predict_task = predict_op(preprocess_task.outputs['processed_test_dataset'], train_task.outputs['model'])

之后运行以下命令来把pipeline.py编译为yaml文件并打包

dsl-compile --py pipeline.py --output pipeline.tar.gz

运行Pipeline

在Kubeflow的dashboard中创建一个Pipeline，上传之前打包的pipeline文件，然后创建一个运行即可。Pipeline运行后的结果如下：



模型的性能指标如下：