【Python机器学习系列】针对特定数据构建管道流水线进行机器学习预测（案例+源码）

这是我的第305篇原创文章。

一、引言

       机器学习项目中有可以自动化的标准工作流程。在 Python scikit-learn 中，管道有助于明确定义和自动化这些工作流程。使用pipeline后，我们每一步的输出都会自动的作为下一个的输入。一套完整的机器学习应用流程如下：

       其中特征工程（本质是多个转换器）以及模型的构建和训练（本质是一个估计器）可以通过一个流水线管道来实现。根据要解决的问题，自定义转换器可以帮助克服预定义转换器无法处理的困难。同时也能够十分灵活的处理列。下面将会使用一些例子来说明它的强大之处。这个例子的目的是让读者熟悉他们可以用来创建转换器和管道的工具，这将使他们能够尽可能高效地以任何方式针对任何数据集设计和预处理特征。

二、自定义转换器

2.1 数据提取转换器

该数据集包含分类和数值自变量的混合，它们必须通过单独的管道进行适当的预处理，然后我们将它们组合在一起。因此，两个管道的第一步都必须是提取需要进行预处理的适当列。下面是名为 FeatureSelector 的自定义转换器的代码。此构造函数的转换方法仅提取并返回 Pandas 数据集，其中仅包含其名称在其初始化期间作为参数传递给它的那些列。

class FeatureSelector(BaseEstimator, TransformerMixin):# 构造函数，传递列参数用于列抽取# 可以加入一些打印信息，看看执行的流程def __init__(self, feature_names):self.feature_names = feature_names# print('FeatureSelector init exce...')# 返回对象本身def fit(self, X, y=None):# print('FeatureSelector fit exce...')return self# 我们需要重写transform方法def transform(self, X, y=None):# print('FeatureSelector transform exce...')return X[self.feature_names]

2.2 分类管道中的自定义转化器

自定义转换器将在分类管道中处理的功能列表：

• • date ：此列中的日期格式为“YYYYMMDDT000000”，必须经过清理和处理才能以任何有意义的方式使用。这个转换器的构造函数将允许我们为参数“use_dates”指定一个值列表，具体取决于我们是要为年、月和日或这些值的某种组合创建一个单独的列，还是简单地完全忽略该列传入一个空列表。通过不对该功能的规范进行硬编码，我们可以随时尝试不同的值组合，而无需重写代码。

  • waterfront：房子是否是近邻海滨。转换为二进制 - 是或否

  • view : 房子被查看了多少次。大多数值是 0。其余的值在 1 和 4 之间分布非常稀疏。转换为二进制 — 是或否

  • yr_renovated : 房子翻新的年份。大部分值都是 0，大概是从不，而其余的在几年之间的分布非常稀薄。转换为二进制 - 是或否

# 构建自定义的分类列Transformer
class CategoricalTransformer(BaseEstimator, TransformerMixin):def __init__(self, use_dates=['year', 'month', 'day']):self._use_dates = use_dates# print('CategoricalTransformer init exce...')def fit(self, X, y=None):return selfdef get_year(self, obj):return str(obj)[:4]def get_month(self, obj):return str(obj)[4:6]def get_day(self, obj):return str(obj)[6:8]def create_binary(self, obj):if obj == 0:return 'No'else:return 'Yes'def transform(self, X, y=None):# print('CategoricalTransformer transform exce...')for spec in self._use_dates:exec("X.loc[:,'{}'] = X['date'].apply(self.get_{})".format(spec, spec))X = X.drop(columns=['date'], axis=1)X.loc[:, 'view'] = X['view'].apply(self.create_binary)X.loc[:, 'waterfront'] = X['waterfront'].apply(self.create_binary)X.loc[:, 'yr_renovated'] = X['yr_renovated'].apply(self.create_binary)return X.values

2.3 数值管道中的自定义转换器

自定义数值转换器在管道中处理的功能列表：

• • bedrooms：房子里的卧室数量。照原样通过。

  • bathrooms: 房子里的浴室数量。这个转换器的构造函数将有一个参数“bath_per_bead”，它接受一个布尔值。如果为 True，则构造函数将通过计算浴室/卧室来创建一个新列，以计算每间卧室的浴室数量并删除原始浴室列。如果为 False，则它只会按原样通过浴室列。

  • sqft_living ：房屋居住面积的平方英尺。照原样通过。

  • sqft_lot ：以平方英尺为单位的地块总面积。照原样通过。

  • floors：房屋的楼层数。照原样通过。

  • condition ：描述房屋状况的离散变量，取值范围为 1–5。照原样通过。

  • grade：给予住房单元的总体等级，基于金县分级系统，值从 1 到 13。照原样通过。

  • sqft_basement ：房屋地下室的大小（如果有），以平方英尺为单位。0 表示没有地下室的房屋。照原样通过。

  • yr_built : 房子的建造年份。这个转换器的构造函数将有另一个参数“years_old”，它也接受一个布尔值。如果为 True，则构造函数将通过从 2019 年减去房屋建造年份来计算 2019 年房屋的年龄来创建一个新列，并删除原始的 yr_built 列。如果为 False，则它只会按原样传递 yr_built 列

# 自定义数值列的转换处理器
class NumericalTransformer(BaseEstimator, TransformerMixin):# 构造函数，bath_per_bed ,years_old控制是否计算卧室和时间处理def __init__(self, bath_per_bed=True, years_old=True):self._bath_per_bed = bath_per_bedself._years_old = years_old# 直接返回转换器本身def fit(self, X, y=None):return self# 我们编写的自定义变换方法创建了上述特征并删除了冗余特征def transform(self, X, y=None):if self._bath_per_bed:# 创建新列X.loc[:, 'bath_per_bed'] = X['bathrooms'] / X['bedrooms']# 删除冗余列X.drop('bathrooms', axis=1)if self._years_old:# 创建新列X.loc[:, 'years_old'] = 2019 - X['yr_built']# 删除冗余列X.drop('yr_built', axis=1)# 将数据集中的任何无穷大值转换为 NanX = X.replace([np.inf, -np.inf], np.nan)# 返回一个 numpy 数组return X.values

三、实现过程

3.1 读取数据

data = pd.read_csv(r'data.csv')
df = pd.DataFrame(data)
print(df.head())

df：

3.2 划分数据集

X = data.drop('price', axis=1)
y = data['price'].values
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

3.3 特征分类传递

传递分类管道的分类特征：

categorical_features = ['date', 'waterfront', 'view', 'yr_renovated']

传递数值管道的数值特征：

numerical_features = ['bedrooms', 'bathrooms', 'sqft_living', 'sqft_lot', 'floors', 'condition', 'grade', 'sqft_basement', 'yr_built']

3.4 管道流水线定义

定义分类管道中的步骤：

categorical_pipeline = Pipeline(steps=[('cat_selector', FeatureSelector(categorical_features)),('cat_transformer', CategoricalTransformer()),('one_hot_encoder', OneHotEncoder(sparse=False))])

定义数值管道中的步骤：

numerical_pipeline = Pipeline(steps=[('num_selector', FeatureSelector(numerical_features)),('num_transformer', NumericalTransformer()),('imputer', SimpleImputer(strategy='median')),('std_scaler', StandardScaler())])

3.5 管道组合

使用FeatureUnion将数值和分类管道水平组合成一个完整大管道：

full_pipeline = FeatureUnion(transformer_list=[('categorical_pipeline', categorical_pipeline),('numerical_pipeline', numerical_pipeline)])

3.6 建立完整管道

完整管道将估算器作为最后一步：

full_pipeline_m = Pipeline(steps=[('full_pipeline', full_pipeline),('model', LinearRegression())])

3.7 管道流水线对训练集进行特征处理和训练

full_pipeline_m.fit(X_train, y_train)

3.8 管道流水线对测试集进行特征处理和预测

y_pred = full_pipeline_m.predict(X_test)
print(y_pred)

y_pred：

作者简介：

读研期间发表6篇SCI数据挖掘相关论文，现在某研究院从事数据算法相关科研工作，结合自身科研实践经历不定期分享关于Python、机器学习、深度学习、人工智能系列基础知识与应用案例。致力于只做原创，以最简单的方式理解和学习，关注我一起交流成长。需要数据集和源码的小伙伴可以关注底部公众号添加作者微信。