Spark MLlib 特征工程(下)

前面我们提到，典型的特征工程包含如下几个环节，即预处理、特征选择、归一化、离散化、Embedding 和向量计算，如下图所示。

在上一讲，我们着重讲解了其中的前 3 个环节，也就是预处理、特征选择和归一化，今天这一讲，咱们继续来说说剩下的离散化、Embedding 与向量计算。

特征工程

离散化：Bucketizer

与归一化一样，离散化也是用来处理数值型字段的。离散化可以把原本连续的数值打散，从而降低原始数据的多样性（Cardinality）。举例来说，“BedroomAbvGr”字段的含义是居室数量，在 train.csv 这份数据样本中，“BedroomAbvGr”包含从 1 到 8 的连续整数。

现在，我们根据居室数量，把房屋粗略地划分为小户型、中户型和大户型

不难发现，“BedroomAbvGr”离散化之后，数据多样性由原来的 8 降低为现在的 3。那么问题来了，原始的连续数据好好的，为什么要对它做离散化呢？离散化的动机，主要在于提升特征数据的区分度与内聚性，从而与预测标的产生更强的关联。

就拿“BedroomAbvGr”来说，我们认为一居室和两居室对于房价的影响差别不大，同样，三居室和四居室之间对于房价的影响，也是微乎其微。但是，小户型与中户型之间，以及中户型与大户型之间，房价往往会出现跃迁的现象。换句话说，相比居室数量，户型的差异对于房价的影响更大、区分度更高。因此，把“BedroomAbvGr”做离散化处理，目的在于提升它与预测标的之间的关联性。

那么，在 Spark MLlib 的框架下，离散化具体该怎么做呢？与其他环节一样，Spark MLlib 提供了多个离散化函数，比如 Binarizer、Bucketizer 和 QuantileDiscretizer。我们不妨以 Bucketizer 为代表，结合居室数量“BedroomAbvGr”这个字段，来演示离散化的具体用法。

// 原始字段
val fieldBedroom: String = "BedroomAbvGrInt"
// 包含离散化数据的目标字段
val fieldBedroomDiscrete: String = "BedroomDiscrete"
// 指定离散区间，分别是[负无穷, 2]、[3, 4]和[5, 正无穷]
val splits: Array[Double] = Array(Double.NegativeInfinity, 3, 5, Double.PositiveInfinity)
 
import org.apache.spark.ml.feature.Bucketizer
 
// 定义并初始化Bucketizer
val bucketizer = new Bucketizer()
// 指定原始列
.setInputCol(fieldBedroom)
// 指定目标列
.setOutputCol(fieldBedroomDiscrete)
// 指定离散区间
.setSplits(splits)
 
// 调用transform完成离散化转换
engineeringData = bucketizer.transform(engineeringData)

不难发现，Spark MLlib 提供的特征处理函数，在用法上大同小异。首先，我们创建 Bucketizer 实例，然后将数值型字段 BedroomAbvGrInt 作为参数传入 setInputCol，同时使用 setOutputCol 来指定用于保存离散数据的新字段 BedroomDiscrete。

离散化的过程是把连续值打散为离散值，但具体的离散区间如何划分，还需要我们通过在 setSplits 里指定。离散区间由浮点型数组 splits 提供，从负无穷到正无穷划分出了[负无穷, 2]、[3, 4]和[5, 正无穷]这三个区间。最终，我们调用 Bucketizer 的 transform 函数，对 engineeringData 做离散化。

离散化前后的数据对比，如下图所示。

Embedding

实际上，Embedding 是一个非常大的话题，随着机器学习与人工智能的发展，Embedding 的方法也是日新月异、层出不穷。从最基本的热独编码到 PCA 降维，从 Word2Vec 到 Item2Vec，从矩阵分解到基于深度学习的协同过滤，可谓百花齐放、百家争鸣。更有学者提出：“万物皆可 Embedding”。那么问题来了，什么是 Embedding 呢？

Embedding 是个英文术语，如果非要找一个中文翻译对照的话，我觉得“向量化”（Vectorize）最合适。Embedding 的过程，就是把数据集合映射到向量空间，进而把数据进行向量化的过程。这句话听上去有些玄乎，我换个更好懂的说法，Embedding 的目标，就是找到一组合适的向量，来刻画现有的数据集合。

以 GarageType 字段为例，它有 6 个取值，也就是说我们总共有 6 种车库类型。那么对于这 6 个字符串来说，我们该如何用数字化的方式来表示它们呢？毕竟，模型只能消费数值，不能直接消费字符串。

种方法是采用预处理环节的 StringIndexer，把字符串转换为连续的整数，然后让模型去消费这些整数。在理论上，这么做没有任何问题。但从模型的效果出发，整数的表达方式并不合理。为什么这么说呢？

我们知道，连续整数之间，是存在比较关系的，比如 1 < 3，6 > 5，等等。但是原始的字符串之间，比如，“Attchd”与“Detchd”并不存在大小关系，如果强行用 0 表示“Attchd”、用 1 表示“Detchd”，逻辑上就会出现“Attchd”<“Detchd”的悖论。

因此，预处理环节的 StringIndexer，仅仅是把字符串转换为数字，转换得到的数值是不能直接喂给模型做训练。我们需要把这些数字进一步向量化，才能交给模型去消费。那么问题来了，对于 StringIndexer 输出的数值，我们该怎么对他们进行向量化呢？这就要用到 Embedding 了

咱们不妨从最简单的热独编码（One Hot Encoding）开始，去认识 Embedding 并掌握它的基本用法。我们先来说说，热独编码，是怎么一回事。

首先，通过 StringIndexer，我们把 GarageType 的 6 个取值分别映射为 0 到 5 的六个数值。接下来，使用热独编码，我们把每一个数值都转化为一个向量。向量的维度为 6，与原始字段（GarageType）的多样性（Cardinality）保持一致。换句话说，热独编码的向量维度，就是原始字段的取值个数。

仔细观察上图的六个向量，只有一个维度取值为 1，其他维度全部为 0。取值为 1 的维度与 StringIndexer 输出的索引相一致。举例来说，字符串“Attchd”被 StringIndexer 映射为 0，对应的热独向量是[1, 0, 0, 0, 0, 0]。向量中索引为 0 的维度取值为 1，其他维度全部取 0。不难发现，热独编码是一种简单直接的 Embedding 方法，甚至可以说是“简单粗暴”。不过，在日常的机器学习开发中，“简单粗暴”的热独编码却颇受欢迎。

在预处理环节，我们已经用 StringIndexer 把非数值字段全部转换为索引字段，接下来，我们再用 OneHotEncoder，把索引字段进一步转换为向量字段。

import org.apache.spark.ml.feature.OneHotEncoder

// 非数值字段对应的目标索引字段，也即StringIndexer所需的“输出列”
// val indexFields: Array[String] = categoricalFields.map(_ + "Index").toArray

// 热独编码的目标字段，也即OneHotEncoder所需的“输出列”
val oheFields: Array[String] = categoricalFields.map(_ + "OHE").toArray

// 循环遍历所有索引字段，对其进行热独编码
for ((indexField, oheField) <- indexFields.zip(oheFields)) {
  val oheEncoder = new OneHotEncoder()
    .setInputCol(indexField)
    .setOutputCol(oheField)
  engineeringData= oheEncoder.fit(engineeringData).transform(engineeringData)
}

可以看到，我们循环遍历所有非数值特征，依次创建 OneHotEncoder 实例。在实例初始化的过程中，我们把索引字段传入给 setInputCol 函数，把热独编码目标字段传递给 setOutputCol 函数。最终通过调用 OneHotEncoder 的 transform，在 engineeringData 之上完成转换。

其实我们也可以OneHotEncoder输入输出接受数组，也就是说我们可以不用使用循环，看起来代码更简洁

val oheEncoder = new OneHotEncoder()
oheEncoder.setInputCols(indexFields).setOutputCols(oheFields)
engineeringData= oheEncoder.fit(engineeringData).transform(engineeringData)

向量计算

向量计算，作为特征工程的最后一个环节，主要用于构建训练样本中的特征向量（Feature Vectors）。在 Spark MLlib 框架下，训练样本由两部分构成

1. 第一部分是预测标的（Label），在“房价预测”的项目中，Label 是房价。
2. 第二部分，就是特征向量，在形式上，特征向量可以看作是元素类型为 Double 的数组。

根据前面的特征工程流程图，我们不难发现，特征向量的构成来源多种多样，比如原始的数值字段、归一化或是离散化之后的数值字段、以及向量化之后的特征字段，等等。

Spark MLlib 在向量计算方面提供了丰富的支持，比如前面介绍过的、用于集成特征向量的 VectorAssembler，用于对向量做剪裁的 VectorSlicer，以元素为单位做乘法的 ElementwiseProduct，等等。灵活地运用这些函数，我们可以随意地组装特征向量，从而构建模型所需的训练样本。

在前面的几个环节中（预处理、特征选择、归一化、离散化、Embedding），我们尝试对数值和非数值类型特征做各式各样的转换，目的在于探索可能对预测标的影响更大的潜在因素。

接下来，我们使用 VectorAssembler 将这些潜在因素全部拼接在一起、构建特征向量，从而为后续的模型训练准备好训练样本。

import org.apache.spark.ml.feature.VectorAssembler
 
/**
入选的数值特征：selectedFeatures
归一化的数值特征：scaledFields
离散化的数值特征：fieldBedroomDiscrete
热独编码的非数值特征：oheFields
*/
 
val assembler = new VectorAssembler()
.setInputCols(selectedFeatures.toArray ++ scaledFields ++ Array(fieldBedroomDiscrete) ++ oheFields)
.setOutputCol("features")
 
engineeringData = assembler.transform(engineeringData)

训练模型

转换完成之后，engineeringData 这个 DataFrame 就包含了一列名为“features”的新字段，这个字段的内容，就是每条训练样本的特征向量。接下来，我们就可以像上一讲那样，通过 setFeaturesCol 和 setLabelCol 来指定特征向量与预测标的，定义出线性回归模型。

// 定义线性回归模型
val lr = new LinearRegression()
.setFeaturesCol("features")
.setLabelCol("SalePriceInt")
.setMaxIter(100)
 
// 训练模型
val lrModel = lr.fit(engineeringData)
 
// 获取训练状态
val trainingSummary = lrModel.summary
// 获取训练集之上的预测误差
println(s"Root Mean Squared Error (RMSE) on train data: ${trainingSummary.rootMeanSquaredError}")

到此为止，我们打通了特征工程所有关卡

你可能会好奇：“这些不同环节的特征处理，真的会对模型效果有帮助吗？毕竟，折腾了半天，我们还是要看模型效果的”。没错，特征工程的最终目的，是调优模型效果。接下来，通过将不同环节输出的训练样本喂给模型，我们来对比不同特征处理方法对应的模型效果。

总结

可以看到，随着特征工程的推进，模型在训练集上的预测误差越来越小，这说明模型的拟合能力越来越强，而这也就意味着，特征工程确实有助于模型性能的提升。