【SparkML实践5】特征转换FeatureTransformers实战scala版

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

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

本章节主要讲转换1

Feature Transformers

Tokenizer（分词）

分词是将文本（如一个句子）拆分成单独词汇（通常是单词）的过程。一个简单的Tokenizer类提供了这项功能。下面的例子展示了如何将句子分割成单词序列。

RegexTokenizer允许基于正则表达式（regex）匹配进行更高级的分词。默认情况下，参数“pattern”（正则表达式，默认值：“\s+”）被用作分隔符来分割输入文本。或者，用户可以将参数“gaps”设置为false，表示正则表达式“pattern”指的是“tokens”，而不是分割间隙，并找到所有匹配的情况作为分词结果。


import org.apache.spark.ml.feature.{
   RegexTokenizer, Tokenizer}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object TokenizerExample {
   
  def main(args: Array[String]): Unit = {
   
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("TokenizerExample")
      .getOrCreate()

    // 数据样本
    val sentenceDataFrame = spark.createDataFrame(Seq(
      (0, "Hi I heard about Spark"),
      (1, "I wish Java could use case classes"),
      (2, "Logistic,regression,models,are,neat")
    )).toDF("id", "sentence")

    val tokenizer = new Tokenizer().setInputCol("sentence").setOutputCol("words")
    val regexTokenizer = new RegexTokenizer()
      .setInputCol("sentence")
      .setOutputCol("words")
      .setPattern("\\W") // 或者 .setPattern("\\w+").setGaps(false)

    val countTokens = udf {
    (words: Seq[String]) => words.length }

    val tokenized = tokenizer.transform(sentenceDataFrame)
    tokenized.select("sentence", "words")
        .withColumn("tokens", countTokens(col("words"))).show(false)

    val regexTokenized = regexTokenizer.transform(sentenceDataFrame)
    regexTokenized.select("sentence", "words")
        .withColumn("tokens", countTokens(col("words"))).show(false)


    spark.stop()
  }
}

StopWordsRemover（停用词）

停用词是那些应该从输入中排除的词，通常是因为这些词出现频率高而且不携带太多意义。

StopWordsRemover接受一个字符串序列作为输入（例如Tokenizer的输出），并丢弃输入序列中的所有停用词。停用词的列表由stopWords参数指定。可以通过调用StopWordsRemover.loadDefaultStopWords(language)获取一些语言的默认停用词列表，可选的语言有“danish”（丹麦语）、“dutch”（荷兰语）、“english”（英语）、“finnish”（芬兰语）、“french”（法语）、“german”（德语）、“hungarian”（匈牙利语）、“italian”（意大利语）、“norwegian”（挪威语）、“portuguese”（葡萄牙语）、“russian”（俄语）、“spanish”（西班牙语）、“swedish”（瑞典语）和“turkish”（土耳其语）。一个布尔参数caseSensitive指示匹配是否应该区分大小写（默认为false）。

示例

假设我们有以下DataFrame，包含列id和raw：

id	raw
0	[I, saw, the, red, balloon]
1	[Mary, had, a, little, lamb]

应用StopWordsRemover，以raw作为输入列，以filtered作为输出列，我们应该得到以下结果：

id	raw	filtered
0	[I, saw, the, red, balloon]	[saw, red, balloon]
1	[Mary, had, a, little, lamb]	[Mary, little, lamb]

在filtered中，停用词“I”、“the”、“had”和“a”已经被过滤掉了。

import org.apache.spark.ml.feature.StopWordsRemover
// $example off$
import org.apache.spark.sql.SparkSession

object StopWordsRemoverExample {
   
  def main(args: Array[String]): Unit = {
   
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("StopWordsRemoverExample")
      .getOrCreate()

    //创建停用词移除器
    val remover = new StopWordsRemover()
      .setInputCol("raw")
      .setOutputCol("filtered")

    val dataSet = spark.createDataFrame(Seq(
      (0, Seq("I", "saw", "the", "red", "balloon")),
      (1, Seq("Mary", "had", "a", "little", "lamb"))
    )).toDF("id", "raw")

    remover.transform(dataSet).show(false)

    spark.stop()
  }
}

n-gram

n-gram是n个连续词条（通常是单词）的序列，其中n是某个整数。NGram类可以用来将输入特征转换成n-grams。

NGram接受一系列字符串作为输入（例如，Tokenizer的输出）。参数n用于确定每个n-gram中的词条数。输出将包括一系列n-grams，其中每个n-gram由n个连续单词组成的以空格分隔的字符串表示。如果输入序列包含的字符串少于n个，则不会产生输出。


import org.apache.spark.ml.feature.NGram
// $example off$
import org.apache.spark.sql.SparkSession

object NGramExample {
   
  def main(args: Array[String]): Unit = {
   
    val spark = SparkSession
      .builder
      .master("local")
      .appName("NGramExample")
      .getOrCreate()

    val wordDataFrame = spark.createDataFrame(Seq(
      (0, Array("Hi", "I", "heard", "about", "Spark")),
      (1, Array("I", "wish", "Java", "could", "use", "case", "classes")),
      (2, Array("Logistic", "regression", "models", "are", "neat"))
    )).toDF("id", "words")
    //NGram 的输出是一个数组，每个元素都是一个 n-gram。
    //如果输入数组的长度小于 n，则不会生成 n-grams。
    //NGram 主要用于文本数据，但理论上它可以用于任何序列化的特征。
    val ngram = new NGram().setN(2).setInputCol("words").setOutputCol("ngrams")

    val ngramDataFrame = ngram.transform(wordDataFrame)
    ngramDataFrame.select("ngrams").show(false)

    spark.stop()
  }
}

Binarizer

二值化是将数值特征阈值化为二进制（0/1）特征的过程。

二值化器（Binarizer）接受常见的参数inputCol和outputCol，以及用于二值化的阈值。特征值大于阈值的将被二值化为1.0；等于或小于阈值的值将被二值化为0.0。对于inputCol，既支持Vector类型也支持Double类型。


import org.apache.spark.ml.feature.Binarizer
// $example off$
import org.apache.spark.sql.SparkSession

object BinarizerExample {
   
  def main(args: Array[String]): Unit = {
   
    val spark = SparkSession
      .builder
      .master("local")
      .appName("BinarizerExample")
      .getOrCreate()

    val data = Array((0, 0.1), (1, 0.8), (2, 0.2))
    val dataFrame = spark.createDataFrame(data).toDF("id", "feature")

    val binarizer: Binarizer = new Binarizer()
      .setInputCol("feature")
      .setOutputCol("binarized_feature")
      .setThreshold(0.5)

    val binarizedDataFrame = binarizer.transform(dataFrame)

    println(s"Binarizer output with Threshold = ${
     binarizer.getThreshold}")
    binarizedDataFrame.show()

    spark.stop()
  }
}

PCA

PCA（主成分分析）是一种统计程序，它使用正交变换将一组可能相关的变量的观测值转换为一组线性无关的变量值，这些变量值称为主成分。PCA类训练一个模型，使用PCA将向量投影到低维空间中。下面的例子展示了如何将5维特征向量投影到3维主成分中。

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

object PCAExample {
   
  def main(args: Array[String]): Unit = {
   
    val spark = SparkSession
      .builder
      .master("local")
      .appName("PCAExample")
      .getOrCreate()

    val data = Array(
      Vectors.sparse(5, Seq((1, 1.0), (3, 7.0))),
      Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),
      Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0)
    )
    val df = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
    // 初始化PCA，设置降维到3维
    val pca = new PCA()
      .setInputCol("features")
      .setOutputCol("pcaFeatures")
      .setK(3)
      .fit(df)

    val result = pca.transform(df).select("pcaFeatures")
    result.show(false)

    spark.stop()
  }
}

PolynomialExpansion

多项式展开是将特征扩展到多项式空间的过程，这个空间是由原始维度的n次组合构成的。PolynomialExpansion类提供了这个功能。下面的例子展示了如何将特征扩展到3次多项式空间。

import org.apache.spark.ml.feature.PolynomialExpansion
import org.apache.spark.ml.linalg.Vectors

import org.apache.spark.sql.SparkSession

object PolynomialExpansionExample {
   
  def main(args: Array[String]): Unit = {
   
    val spark = SparkSession
      .builder
      .appName("PolynomialExpansionExample")
      .getOrCreate()

    val data = Array(
      Vectors.dense(2.0, 1.0),
      Vectors.dense(0.0, 0.0),
      Vectors.dense(3.0, -1.0)
    )
    val df = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")

    val polyExpansion = new PolynomialExpansion()
      .setInputCol("features")
      .setOutputCol("polyFeatures")
      .setDegree(3)

    val polyDF = polyExpansion.transform(df)
    polyDF.show(false)

    spark.stop()
  }
}

Discrete Cosine Transform (DCT)

离散余弦变换（Discrete Cosine Transform，简称DCT）将时域中长度为N的实数序列转换为频域中另一长度为N的实数序列。一个DCT类提供了这个功能，实现了DCT-II，并通过1/2的平方根来缩放结果。转换后的序列不会应用位移（例如，变换序列的第0个元素是第0个DCT系数，而不是第N/2个）。


import org.apache.spark.ml.feature.DCT
import org.apache.spark.ml.linalg.Vectors
// $example off$
import org.apache.spark.sql.SparkSession

object DCTExample {
   
  /**
   * 离散余弦变换（Discrete Cosine Transform，简称DCT）是一种用于信号和图像处理的变换，它类似于离散傅立叶变换（Discrete Fourier Transform，简称DFT），但只使用实数。DCT在多个领域非常有用，尤其是在压缩技术中，比如JPEG图像压缩就广泛使用了DCT。
   */
  def main(args: Array[String]): Unit = {
   
    val spark = SparkSession
      .builder
      .appName("DCTExample")
      .getOrCreate()

    val data = Seq(
      Vectors.dense(0.0, 1.0, -2.0, 3.0),
      Vectors.dense(-1.0, 2.0, 4.0, -7.0),
      Vectors.dense(14.0, -2.0, -5.0, 1.0))

    val df = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")

    val dct = new DCT()
      .setInputCol("features")
      .setOutputCol("featuresDCT")
      .setInverse(false)

    val dctDf = dct.transform(df)
    dctDf.select("featuresDCT").show(false)

    spark.stop()
  }
}

StringIndexer

StringIndexer 将包含标签的字符串列编码为标签索引列。StringIndexer 可以编码多个列。索引位于 [0, numLabels) 范围内，并支持四种排序选项：“frequencyDesc”（按标签频率降序，频率最高的标签分配索引0）、“frequencyAsc”（按标签频率升序，频率最低的标签分配索引0）、“alphabetDesc”（按字母表降序）和“alphabetAsc”（按字母表升序，默认为“frequencyDesc”）。注意，在“frequencyDesc”/“frequencyAsc”下，如果频率相同，字符串会进一步按字母表排序。

如果用户选择保留未见过的标签，这些标签将被放置在索引 numLabels 处。如果输入列是数值型，我们会将其转换为字符串并对这些字符串值进行索引。当下游的管道组件如 Estimator 或 Transformer 使用这个字符串索引的标签时，你必须将该组件的输入列设置为这个字符串索引的列名。在许多情况下，你可以使用 setInputCol 来设置输入列。

示例

假设我们有以下包含 id 和 category 列的 DataFrame：

id	category
0	a
1	b
2	c
3	a
4	a
5	c

category 是一个字符串列，包含三个标签：“a”、“b”和“c”。应用 StringIndexer，并将 category 设置为输入列，categoryIndex 设置为输出列，我们应该得到以下结果：

id	category	categoryIndex
0	a	0.0
1	b	2.0
2	c	1.0
3	a	0.0
4	a	0.0
5	c	1.0

“a” 获得索引0，因为它是最频繁的，其次是“c”获得索引1，“b”获得索引2。

此外，关于当你在一个数据集上训练了 StringIndexer，然后用它来转换另一个数据集时，StringIndexer 将如何处理未见过的标签，有三种策略：

抛出一个异常（这是默认行为）
完全跳过包含未见标签的行
将未见标签放在一个特殊的额外桶中，索引为 numLabels

让我们回到我们之前的例子，但这次在以下数据集上重用我们之前定义的 StringIndexer：

id	category
0	a
1	b
2	c
3	d
4	e

如果你没有设置 StringIndexer 如何处理未见过的标签，或者将其设置为“error”，那么将会抛出一个异常。然而，如果你之前调用了 setHandleInvalid(“skip”)，那么将会生成以下数据集：

id	category	categoryIndex
0	a	0.0
1	b	2.0
2	c	1.0

请注意，包含“d”或“e”的行不会出现。

如果你调用 setHandleInvalid(“keep”)，那么将会生成以下数据集：

id	category	categoryIndex
0	a	0.0
1	b	2.0
2	c	1.0
3	d	3.0
4	e	3.0

请注意，包含“d”或“e”的行被映射到索引“3.0”。

import org.apache.spark.ml.feature.StringIndexer
import org.apache.spark.sql.SparkSession

object StringIndexerExample {
   
  def main(args: Array[String]): Unit = {
   
    val spark = SparkSession
      .builder
      .master("local")
      .appName("StringIndexerExample")
      .getOrCreate()

    val df = spark.createDataFrame(
      Seq((0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c"))
    ).toDF("id", "category")

    val indexer = new StringIndexer()
      .setInputCol("category")
      .setOutputCol("categoryIndex")
      .setHandleInvalid("skip") //#  "error" 或 "keep"

    val indexed = indexer.fit(df).transform(df)
    indexed.show()

    spark.stop()
  }
}