【SparkML实践7】特征选择器FeatureSelector

本节介绍了用于处理特征的算法，大致可以分为以下几组：

• 提取（Extraction）：从“原始”数据中提取特征。
• 转换（Transformation）：缩放、转换或修改特征。
• 选择（Selection）：从更大的特征集中选择一个子集。
• 局部敏感哈希（Locality Sensitive Hashing, LSH）：这类算法结合了特征转换的方面与其他算法。

Feature Selectors

VectorSlicer

VectorSlicer 是一个转换器，它接受一个特征向量，并输出一个新的特征向量，该向量包含原始特征的子数组。它用于从向量列中提取特征。

VectorSlicer 接受一个带有指定索引的向量列，然后输出一个新的向量列，其值通过这些索引选择。有两种类型的索引：

1. 整数索引，代表向量中的索引，使用 setIndices() 设置。
2. 字符串索引，代表向量中的特征名称，使用 setNames() 设置。这要求向量列具有 AttributeGroup，因为实现是基于 Attribute 的 name 字段进行匹配的。

整数和字符串规格都是可以接受的。此外，您可以同时使用整数索引和字符串名称。至少必须选择一个特征。不允许有重复的特征，所以选定的索引和名称之间不能有重叠。请注意，如果选择了特征的名称，在遇到空的输入属性时会抛出异常。

输出向量将首先按照给定的顺序排列选定的索引特征，然后按照给定的顺序排列选定的名称特征。

Examples

Suppose that we have a DataFrame with the column userFeatures:

userFeatures	x
[0.0, 10.0, 0.5]

userFeatures 是一个向量列，包含三个用户特征。假设 userFeatures 的第一列全是零，因此我们想要移除它，只选择最后两列。VectorSlicer 通过 setIndices(1, 2) 选择最后两个元素，然后生成一个名为 features 的新向量列：

userFeatures	features
[0.0, 10.0, 0.5]	[10.0, 0.5]

假设我们还有 userFeatures 的潜在输入属性，即 [“f1”, “f2”, “f3”]，那么我们可以使用 setNames(“f2”, “f3”) 来选择它们。

userFeatures	features
[0.0, 10.0, 0.5]	[10.0, 0.5]
[“f1”, “f2”, “f3”]	[“f2”, “f3”]

import org.apache.spark.ml.attribute.{Attribute, AttributeGroup, NumericAttribute}
import org.apache.spark.ml.feature.VectorSlicer
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{Row, SparkSession}import java.util.Arraysobject VectorSlicerExample {def main(args: Array[String]): Unit = {val spark = SparkSession.builder.master("local").appName("VectorSlicerExample").getOrCreate()val data = Arrays.asList(Row(Vectors.sparse(3, Seq((0, -2.0), (1, 2.3)))),Row(Vectors.dense(-2.0, 2.3, 0.0)))val defaultAttr = NumericAttribute.defaultAttrval attrs = Array("f1", "f2", "f3").map(defaultAttr.withName)val attrGroup = new AttributeGroup("userFeatures", attrs.asInstanceOf[Array[Attribute]])val dataset = spark.createDataFrame(data, StructType(Array(attrGroup.toStructField())))val slicer = new VectorSlicer().setInputCol("userFeatures").setOutputCol("features")slicer.setIndices(Array(1)).setNames(Array("f3"))// or slicer.setIndices(Array(1, 2)), or slicer.setNames(Array("f2", "f3"))val output = slicer.transform(dataset)output.show(false)spark.stop()}
}

RFormula

RFormula 通过指定 R 模型公式来选择列。目前我们支持 R 操作符的一个有限子集，包括 ‘~’、‘.’、‘:’、‘+’ 和 ‘-’。基本操作符有：

分隔目标和项

连接项，“+ 0” 表示去除截距
移除一个项，“- 1” 表示去除截距
: 交互作用（数值的乘积，或二值化的类别值）
. 所有列除了目标
假设 a 和 b 是双精度列，我们使用以下简单的例子来说明 RFormula 的效果：
y ~ a + b 表示模型 y ~ w0 + w1 * a + w2 * b，其中 w0 是截距，w1、w2 是系数。
y ~ a + b + a:b - 1 表示模型 y ~ w1 * a + w2 * b + w3 * a * b，其中 w1、w2、w3 是系数。
RFormula 生成一个特征向量列和一个双精度或字符串列的标签。就像在 R 中用于线性回归的公式一样，数值列将被转换为双精度数。至于字符串输入列，它们首先会通过 StringIndexer 转换，使用由 stringOrderType 确定的顺序，并且在排序后的最后一个类别会被丢弃，然后双精度数将被进行独热编码。

假设有一个包含值 {‘b’, ‘a’, ‘b’, ‘a’, ‘c’, ‘b’} 的字符串特征列，我们设置 stringOrderType 来控制编码：

ChiSqSelector

ChiSqSelector 代表卡方特征选择。它作用于带有类别特征的标记数据。ChiSqSelector 使用卡方独立性检验来决定选择哪些特征。它支持五种选择方法：numTopFeatures、percentile、fpr、fdr、fwe：

1. numTopFeatures 根据卡方检验选择固定数量的顶级特征。这类似于选择具有最高预测能力的特征。
2. percentile 与 numTopFeatures 类似，但它选择所有特征的一定比例，而不是固定数量。
3. fpr 选择所有 p 值低于阈值的特征，从而控制选择的假阳性率。
4. fdr 使用 Benjamini-Hochberg 程序选择所有假发现率低于阈值的特征。
5. fwe 选择所有 p 值低于阈值的特征。阈值通过 1/numFeatures 缩放，从而控制选择的家族错误率。
6. 默认情况下，选择方法为 numTopFeatures，且默认的顶级特征数量设置为 50。用户可以使用 setSelectorType 选择一个选择方法。

示例

假设我们有一个 DataFrame，它包含列 id、features 和 clicked，clicked 被用作我们要预测的目标：

id	features	clicked
7	[0.0, 0.0, 18.0, 1.0]	1.0
8	[0.0, 1.0, 12.0, 0.0]	0.0
9	[1.0, 0.0, 15.0, 0.1]	0.0

如果我们使用 ChiSqSelector 并设置 numTopFeatures = 1，那么根据我们的标签 clicked，我们特征中的最后一列将被选为最有用的特征：

id	features	clicked	selectedFeatures
7	[0.0, 0.0, 18.0, 1.0]	1.0	[1.0]
8	[0.0, 1.0, 12.0, 0.0]	0.0	[0.0]
9	[1.0, 0.0, 15.0, 0.1]	0.0	[0.1]

import org.apache.spark.ml.feature.ChiSqSelector
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.SparkSessionobject ChiSqSelectorExample {def main(args: Array[String]): Unit = {val spark = SparkSession.builder.master("local").appName("ChiSqSelectorExample").getOrCreate()import spark.implicits._val data = Seq((7, Vectors.dense(0.0, 0.0, 18.0, 1.0), 1.0),(8, Vectors.dense(0.0, 1.0, 12.0, 0.0), 0.0),(9, Vectors.dense(1.0, 0.0, 15.0, 0.1), 0.0))val df = spark.createDataset(data).toDF("id", "features", "clicked")val selector = new ChiSqSelector().setNumTopFeatures(1).setFeaturesCol("features").setLabelCol("clicked").setOutputCol("selectedFeatures")val result = selector.fit(df).transform(df)println(s"ChiSqSelector output with top ${selector.getNumTopFeatures} features selected")result.show()spark.stop()}
}

UnivariateFeatureSelector

单变量特征选择器（UnivariateFeatureSelector）可以操作具有类别型/连续型标签的类别型/连续型特征。用户可以设置特征类型（featureType）和标签类型（labelType），Spark会根据指定的特征类型和标签类型选择使用的评分函数。

特征类型	标签类型	评分函数
categorical（类别型）	categorical	chi-squared (chi2)
continuous	categorical	ANOVATest (f_classif)
continuous	continuous	F-value (f_regression)

它支持五种选择模式：numTopFeatures、percentile、fpr、fdr、fwe：

1. numTopFeatures 选择固定数量的最优特征。
2. percentile 类似于numTopFeatures，但它选择所有特征的一定比例，而不是固定数量。
3. fpr 选择所有p值低于阈值的特征，从而控制选择的假阳性率。
4. fdr 使用Benjamini-Hochberg程序选择所有假发现率低于阈值的特征。
5. fwe 选择所有p值低于阈值的特征。阈值通过1/numFeatures进行缩放，从而控制选择的家族误差率。
   默认情况下，选择模式为numTopFeatures，且默认的selectionThreshold设置为50。

示例

假设我们有一个DataFrame，包含列id、features和label，label是我们预测的目标：

id	features	label
1	[1.7, 4.4, 7.6, 5.8, 9.6, 2.3]	3.0
2	[8.8, 7.3, 5.7, 7.3, 2.2, 4.1]	2.0
3	[1.2, 9.5, 2.5, 3.1, 8.7, 2.5]	3.0
4	[3.7, 9.2, 6.1, 4.1, 7.5, 3.8]	2.0
5	[8.9, 5.2, 7.8, 8.3, 5.2, 3.0]	4.0
6	[7.9, 8.5, 9.2, 4.0, 9.4, 2.1]	4.0
如果我们将特征类型设置为连续型，标签类型设置为类别型，且numTopFeatures = 1，则我们的特征中的最后一列被选为最有用的特征：

id	features	label	selectedFeatures
1	[1.7, 4.4, 7.6, 5.8, 9.6, 2.3]	3.0	[2.3]
2	[8.8, 7.3, 5.7, 7.3, 2.2, 4.1]	2.0	[4.1]
3	[1.2, 9.5, 2.5, 3.1, 8.7, 2.5]	3.0	[2.5]
4	[3.7, 9.2, 6.1, 4.1, 7.5, 3.8]	2.0	[3.8]
5	[8.9, 5.2, 7.8, 8.3, 5.2, 3.0]	4.0	[3.0]
6	[7.9, 8.5, 9.2, 4.0, 9.4, 2.1]	4.0	[2.1]

import org.apache.spark.ml.feature.UnivariateFeatureSelector
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.SparkSession/*** An example for UnivariateFeatureSelector.* Run with* {{{* bin/run-example ml.UnivariateFeatureSelectorExample* }}}*/
object UnivariateFeatureSelectorExample {def main(args: Array[String]): Unit = {val spark = SparkSession.builder.appName("UnivariateFeatureSelectorExample").getOrCreate()import spark.implicits._val data = Seq((1, Vectors.dense(1.7, 4.4, 7.6, 5.8, 9.6, 2.3), 3.0),(2, Vectors.dense(8.8, 7.3, 5.7, 7.3, 2.2, 4.1), 2.0),(3, Vectors.dense(1.2, 9.5, 2.5, 3.1, 8.7, 2.5), 3.0),(4, Vectors.dense(3.7, 9.2, 6.1, 4.1, 7.5, 3.8), 2.0),(5, Vectors.dense(8.9, 5.2, 7.8, 8.3, 5.2, 3.0), 4.0),(6, Vectors.dense(7.9, 8.5, 9.2, 4.0, 9.4, 2.1), 4.0))val df = spark.createDataset(data).toDF("id", "features", "label")val selector = new UnivariateFeatureSelector().setFeatureType("continuous").setLabelType("categorical").setSelectionMode("numTopFeatures").setSelectionThreshold(1).setFeaturesCol("features").setLabelCol("label").setOutputCol("selectedFeatures")val result = selector.fit(df).transform(df)println(s"UnivariateFeatureSelector output with top ${selector.getSelectionThreshold}" +s" features selected using f_classif")result.show()spark.stop()}
}

VarianceThresholdSelector

VarianceThresholdSelector 是一个选择器，用于移除低方差特征。那些样本方差不大于 varianceThreshold 的特征将被移除。如果没有设置 varianceThreshold，默认值为 0，这意味着只有方差为 0 的特征（即在所有样本中具有相同值的特征）将被移除。

示例

假设我们有一个 DataFrame，它包含列 id 和 features，这些特征用作我们要预测的目标：

id	features
1	[6.0, 7.0, 0.0, 7.0, 6.0, 0.0]
2	[0.0, 9.0, 6.0, 0.0, 5.0, 9.0]
3	[0.0, 9.0, 3.0, 0.0, 5.0, 5.0]
4	[0.0, 9.0, 8.0, 5.0, 6.0, 4.0]
5	[8.0, 9.0, 6.0, 5.0, 4.0, 4.0]
6	[8.0, 9.0, 6.0, 0.0, 0.0, 0.0]

这6个特征的样本方差分别为16.67、0.67、8.17、10.17、5.07和11.47。如果我们使用VarianceThresholdSelector并设置varianceThreshold = 8.0，那么方差小于等于8.0的特征将被移除：

id	features	selectedFeatures
1	[6.0, 7.0, 0.0, 7.0, 6.0, 0.0]	[6.0,0.0,7.0,0.0]
2	[0.0, 9.0, 6.0, 0.0, 5.0, 9.0]	[0.0,6.0,0.0,9.0]
3	[0.0, 9.0, 3.0, 0.0, 5.0, 5.0]	[0.0,3.0,0.0,5.0]
4	[0.0, 9.0, 8.0, 5.0, 6.0, 4.0]	[0.0,8.0,5.0,4.0]
5	[8.0, 9.0, 6.0, 5.0, 4.0, 4.0]	[8.0,6.0,5.0,4.0]
6	[8.0, 9.0, 6.0, 0.0, 0.0, 0.0]	[8.0,6.0,0.0,0.0]

Locality Sensitive Hashing

局部敏感哈希（LSH）是一类重要的哈希技术，通常用于大数据集的聚类、近似最近邻搜索和异常值检测。

LSH的基本思想是使用一族函数（“LSH族”）将数据点哈希到桶中，使得彼此接近的数据点有很高的概率落在同一个桶里，而彼此距离较远的数据点则很可能落在不同的桶中。一个LSH族正式定义如下。

在一个度量空间（M, d）中，其中M是一个集合，d是M上的一个距离函数，一个LSH族是一族满足以下性质的函数h：
∀p,q∈M,
d(p,q)≤r1⇒Pr(h§=h(q))≥p1
d(p,q)≥r2⇒Pr(h§=h(q))≤p2
这样的LSH族称为(r1, r2, p1, p2)-敏感的。

在Spark中，不同的LSH族在不同的类中实现（例如，MinHash），并且每个类中都提供了特征转换、近似相似性连接和近似最近邻搜索的API。

在LSH中，我们定义一个假正例为一对距离较远的输入特征（满足d(p,q)≥r2）被哈希到同一个桶中，我们定义一个假反例为一对接近的特征（满足d(p,q)≤r1）被哈希到不同的桶中。

LSH Operations

我们描述了LSH可用于的主要操作类型。一个训练好的LSH模型具有这些操作的各自方法。

Feature Transformation

特征转换是添加哈希值作为新列的基本功能。这对于降维很有用。用户可以通过设置inputCol和outputCol来指定输入和输出列的名称。

LSH还支持多个LSH哈希表。用户可以通过设置numHashTables来指定哈希表的数量。这也用于近似相似性连接和近似最近邻搜索中的OR放大。增加哈希表的数量将提高精度，但也会增加通信成本和运行时间。

outputCol的类型是Seq[Vector]，其中数组的维度等于numHashTables，向量的维度目前设置为1。在未来的版本中，我们将实现AND放大，以便用户可以指定这些向量的维度。

Approximate Similarity Join

近似相似性连接接受两个数据集，并近似返回数据集中距离小于用户定义阈值的行对。近似相似性连接支持连接两个不同的数据集和自连接。自连接会产生一些重复的对。

近似相似性连接接受转换过的和未转换过的数据集作为输入。如果使用未转换的数据集，它将自动被转换。在这种情况下，哈希签名将作为outputCol创建。

在连接的数据集中，可以在datasetA和datasetB中查询原始数据集。输出数据集中将添加一个距离列，以显示返回的每对行之间的真实距离。

Approximate Nearest Neighbor Search

近似最近邻搜索接受一个数据集（特征向量集）和一个键（单个特征向量），它近似返回数据集中最接近该向量的指定数量的行。

近似最近邻搜索接受转换过的和未转换过的数据集作为输入。如果使用未转换的数据集，它将自动被转换。在这种情况下，哈希签名将作为outputCol创建。

输出数据集中将添加一个距离列，以显示每个输出行与搜索键之间的真实距离。

注意：当哈希桶中没有足够的候选者时，近似最近邻搜索将返回少于k行。

LSH Algorithms

import org.apache.spark.ml.feature.BucketedRandomProjectionLSH
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.colval dfA = spark.createDataFrame(Seq((0, Vectors.dense(1.0, 1.0)),(1, Vectors.dense(1.0, -1.0)),(2, Vectors.dense(-1.0, -1.0)),(3, Vectors.dense(-1.0, 1.0))
)).toDF("id", "features")val dfB = spark.createDataFrame(Seq((4, Vectors.dense(1.0, 0.0)),(5, Vectors.dense(-1.0, 0.0)),(6, Vectors.dense(0.0, 1.0)),(7, Vectors.dense(0.0, -1.0))
)).toDF("id", "features")val key = Vectors.dense(1.0, 0.0)val brp = new BucketedRandomProjectionLSH().setBucketLength(2.0).setNumHashTables(3).setInputCol("features").setOutputCol("hashes")val model = brp.fit(dfA)// Feature Transformation
println("The hashed dataset where hashed values are stored in the column 'hashes':")
model.transform(dfA).show()// Compute the locality sensitive hashes for the input rows, then perform approximate
// similarity join.
// We could avoid computing hashes by passing in the already-transformed dataset, e.g.
// `model.approxSimilarityJoin(transformedA, transformedB, 1.5)`
println("Approximately joining dfA and dfB on Euclidean distance smaller than 1.5:")
model.approxSimilarityJoin(dfA, dfB, 1.5, "EuclideanDistance").select(col("datasetA.id").alias("idA"),col("datasetB.id").alias("idB"),col("EuclideanDistance")).show()// Compute the locality sensitive hashes for the input rows, then perform approximate nearest
// neighbor search.
// We could avoid computing hashes by passing in the already-transformed dataset, e.g.
// `model.approxNearestNeighbors(transformedA, key, 2)`
println("Approximately searching dfA for 2 nearest neighbors of the key:")
model.approxNearestNeighbors(dfA, key, 2).show()

MinHash for Jaccard Distance

MinHash是一种用于Jaccard距离的LSH族，输入特征是自然数集合。两个集合的Jaccard距离由它们交集和并集的基数定义：
d(A, B) = 1 - |A ∩ B| / |A ∪ B|
MinHash对集合中的每个元素应用一个随机哈希函数g，并取所有哈希值的最小值：
h(A) = min_{a∈A}(g(a))

MinHash的输入集合表示为二进制向量，向量索引代表元素本身，向量中的非零值表示集合中该元素的存在。尽管支持密集和稀疏向量，但通常推荐使用稀疏向量以提高效率。例如，Vectors.sparse(10, Array[(2, 1.0), (3, 1.0), (5, 1.0)])表示空间中有10个元素。这个集合包含元素2、元素3和元素5。所有非零值都被视为二进制“1”值。

注意：空集不能通过MinHash转换，这意味着任何输入向量必须至少有一个非零条目。

import org.apache.spark.ml.feature.MinHashLSH
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.colval dfA = spark.createDataFrame(Seq((0, Vectors.sparse(6, Seq((0, 1.0), (1, 1.0), (2, 1.0)))),(1, Vectors.sparse(6, Seq((2, 1.0), (3, 1.0), (4, 1.0)))),(2, Vectors.sparse(6, Seq((0, 1.0), (2, 1.0), (4, 1.0))))
)).toDF("id", "features")val dfB = spark.createDataFrame(Seq((3, Vectors.sparse(6, Seq((1, 1.0), (3, 1.0), (5, 1.0)))),(4, Vectors.sparse(6, Seq((2, 1.0), (3, 1.0), (5, 1.0)))),(5, Vectors.sparse(6, Seq((1, 1.0), (2, 1.0), (4, 1.0))))
)).toDF("id", "features")val key = Vectors.sparse(6, Seq((1, 1.0), (3, 1.0)))val mh = new MinHashLSH().setNumHashTables(5).setInputCol("features").setOutputCol("hashes")val model = mh.fit(dfA)// Feature Transformation
println("The hashed dataset where hashed values are stored in the column 'hashes':")
model.transform(dfA).show()// Compute the locality sensitive hashes for the input rows, then perform approximate
// similarity join.
// We could avoid computing hashes by passing in the already-transformed dataset, e.g.
// `model.approxSimilarityJoin(transformedA, transformedB, 0.6)`
println("Approximately joining dfA and dfB on Jaccard distance smaller than 0.6:")
model.approxSimilarityJoin(dfA, dfB, 0.6, "JaccardDistance").select(col("datasetA.id").alias("idA"),col("datasetB.id").alias("idB"),col("JaccardDistance")).show()// Compute the locality sensitive hashes for the input rows, then perform approximate nearest
// neighbor search.
// We could avoid computing hashes by passing in the already-transformed dataset, e.g.
// `model.approxNearestNeighbors(transformedA, key, 2)`
// It may return less than 2 rows when not enough approximate near-neighbor candidates are
// found.
println("Approximately searching dfA for 2 nearest neighbors of the key:")
model.approxNearestNeighbors(dfA, key, 2).show()