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

假设我们还有 userFeatures 的潜在输入属性，即 [“f1”, “f2”, “f3”]，那么我们可以使用 setNames(“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.Arrays

object 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.defaultAttr
    val 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 被用作我们要预测的目标：

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

import org.apache.spark.ml.feature.ChiSqSelector
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.SparkSession

object 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会根据指定的特征类型和标签类型选择使用的评分函数。

它支持五种选择模式：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是我们预测的目标：

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，这些特征用作我们要预测的目标：

这6个特征的样本方差分别为16.67、0.67、8.17、10.17、5.07和11.47。如果我们使用VarianceThresholdSelector并设置varianceThreshold = 8.0，那么方差小于等于8.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.col

val 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.col

val 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()

代码007普通

打赏 收藏 海报 链接