Spark ML 4.特征提取

spark

发布日期: 2018-12-27

文章字数: 3.5k

阅读时长: 15 分

阅读次数:

1. 特征处理介绍

特征处理主要分三部分：

特征提取：从原始数据中提取特征
特征转换：特征的维度、特征的转化、特征的修改
特征选取：从大规模特征中选取一个子集

Spark 特征提取提供三种算法：分别是 TF-IDF、 Word2Vec 以及 CountVectorizer

2. TF-IDF（词频－逆向文件频率）

2.1 算法介绍

类别：transformer【转换器】

词频 – 逆向文件频率（TF-IDF（HashingTF and IDF））是一种在本文挖掘中广泛使用的特征向量化方法，它可以体现一个文档中词语在语料库中的重要程度。

t	词语
d	文档
D	语料库
TF(t,d)	词频：是词语t 在文档d中出现的次数
DF(t,D)	文件频率：包含词语t的文档个数

如果我们只使用词频来衡量重要性，很容易过度强调在文档中经常出现，却没有太多实际信息的词语，比如“a”，“the”以及“of”。
如果一个词语经常出现在语料库中，意味着它并不能很好的对文档进行区分。
TF-IDF就是在数值化文档信息，衡量词语能提供多少信息以区分文档。
其定义为：$IDF(t,D) = log \frac{|D|+1}{DF(t,D)+1} = \frac{文档总数+1}{文档频率+1}$
- 此处 |D| 是语料库中总的文档数。
- 公式中使用log函数，当词出现在所有文档中时，它的IDF值变为0。
- 加1是为了避免分母为0的情况。
TF-IDF 度量值表示为：$TFIDF(t,d,D) = TF(t,d) · IDF(t,D)$

在Spark ML库中，TF-IDF被分成两部分：TF (+hashing) 和 IDF。

TF：HashingTF 是一个Transformer，在文本处理中，接收词条的集合然后把这些集合转化成固定长度的特征向量。这个算法在哈希的同时会统计各个词条的词频。

IDF：IDF是一个Estimator，在一个数据集上应用它的fit()方法，产生一个IDFModel。该IDFModel 接收特征向量（由HashingTF产生），然后计算每一个词在文档中出现的频次。IDF会减少那些在语料库中出现频率较高的词的权重。

Spark.mllib 中实现词频率统计使用特征hash的方式，原始特征通过hash函数，映射到一个索引值。后面只需要统计这些索引值的频率，就可以知道对应词的频率。这种方式避免设计一个全局1对1的词到索引的映射，这个映射在映射大量语料库时需要花费更长的时间。但需要注意，通过hash的方式可能会映射到同一个值的情况，即不同的原始特征通过Hash映射后是同一个值。为了降低这种情况出现的概率，我们只能对特征向量升维。i.e., 提高hash表的桶数，默认特征维度是 $2^{20} = 1,048,576$

在下面的代码段中，我们以一组句子开始。首先使用分解器Tokenizer把句子划分为单个词语。对每一个句子（词袋），我们使用HashingTF将句子转换为特征向量，最后使用IDF重新调整特征向量。这种转换通常可以提高使用文本特征的性能。

2.2 代码示例

package hnbian.spark.ml.feature.extractors

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}
import org.apache.spark.sql.SparkSession

/**
  * @author hnbian
  * @ Description
  * @ date 2018/12/26 14:46
  **/
object TF_IDF extends App {
  val conf = new SparkConf().setAppName("TF_IDF")
  //设置master local[4] 指定本地模式开启模拟worker线程数
  conf.setMaster("local[4]")
  //创建sparkContext文件
  val sc = new SparkContext(conf)
  val spark = SparkSession.builder().getOrCreate()
  sc.setLogLevel("Error")
  //开启隐式转换
  import spark.implicits._

  //创建一个DataFrame ，每一个句子代表一个文档。

  val sentenceData = spark.createDataFrame(Seq(
    (0,"I heard about Spark and If love Spark"),
    (0,"I wish Java could Spark use case classes"),
    (1,"Logistic regression Spark models are neat"),
  )).toDF("label","sentence")

  //查看我们刚刚创建的DataFrame
  sentenceData.show(false)
  /**
    * +-----+------------------------------------+
    * |label|sentence                            |
    * +-----+------------------------------------+
    * |0    |I heard about Spark and I love Spark|
    * |0    |I wish Java could use case classes  |
    * |1    |Logistic regression models are neat |
    * +-----+------------------------------------+
    */

  //使用分词器对句子进行分词
  //1.创建分词器 （转换器）
  val tokenizer = new Tokenizer()
    .setInputCol("sentence")
    .setOutputCol("words")
  //2.调用转换器的transform()方法 对句子进行分词
  val wordsData = tokenizer.transform(sentenceData)
  //查看分词器执行结果
  wordsData.show(false)

  /**
    * +-----+------------------------------------+---------------------------------------------+
    * |label|sentence                            |words                                        |
    * +-----+------------------------------------+---------------------------------------------+
    * |0    |I heard about Spark and I love Spark|[i, heard, about, spark, and, i, love, spark]|
    * |0    |I wish Java could use case classes  |[i, wish, java, could, use, case, classes]   |
    * |1    |Logistic regression models are neat |[logistic, regression, models, are, neat]    |
    * +-----+------------------------------------+---------------------------------------------+
    */

  //使用HashingTF 调用transform()方法，把分出来的词转换成特征向量
  //1.定义HashingTF
  val hashingTF = new HashingTF()
    .setInputCol("words")
    .setOutputCol("newFeatures")
    .setNumFeatures(2000) // 设置哈希表桶数为2000 （默认2^20 = 1,048,576）

  //2.调用HashingTF.tronsform()方法 转换特征向量
  val featurizedData = hashingTF.transform(wordsData)

  //查看转换特征向量后的数据
  featurizedData.show(false)

  /**
    * +-----+-----------------------------------------+-------------------------------------------------+------------------------------------------------------------------------------+
    * |label|sentence                                 |words                                            |newFeatures 前面数组是将词转成向量，后面数组是对应下表位置词语在所在文档出现次数  |
    * +-----+-----------------------------------------+-------------------------------------------------+------------------------------------------------------------------------------+
    * |0    |I heard about Spark and If love Spark    |[i, heard, about, spark, and, if, love, spark]   |(2000,[170,240,333,1105,1329,1357,1777],[1.0,1.0,1.0,2.0,1.0,1.0,1.0])        |
    * |0    |I wish Java could Spark use case classes |[i, wish, java, could, spark, use, case, classes]|(2000,[213,342,489,495,1105,1329,1809,1967],[1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0])|
    * |1    |Logistic regression Spark models are neat|[logistic, regression, spark, models, are, neat] |(2000,[286,695,1105,1138,1193,1604],[1.0,1.0,1.0,1.0,1.0,1.0])                |
    * +-----+-----------------------------------------+-------------------------------------------------+------------------------------------------------------------------------------+
    */

  //使用 IDF对单纯词频特征向量进行修正，使其更能体现不同词汇文本的区别能力，
  //IDF 是一个Estimator，调用fit(）方法并将词频传入，然后产生一个IDFModel
  //1.定义一个IDF
  val idf = new IDF().setInputCol("newFeatures").setOutputCol("features")

  //2.调用IDF.fit()方法，产生一个model
  val idfModel = idf.fit(featurizedData)

  //idfModel是一个transformer调用它的transform(）方法可以得到每个单词对应的TF-IDF度量值
  val rescaledData = idfModel.transform(featurizedData)
  rescaledData.show(false)

  /**
    *
    * +-----+-----------------------------------------+-------------------------------------------------+------------------------------------------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
    * |label|sentence                                 |words                                            |newFeatures                                                                   |features                                                                                                                                                                                |
    * +-----+-----------------------------------------+-------------------------------------------------+------------------------------------------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
    * |0    |I heard about Spark and If love Spark    |[i, heard, about, spark, and, if, love, spark]   |(2000,[170,240,333,1105,1329,1357,1777],[1.0,1.0,1.0,2.0,1.0,1.0,1.0])        |(2000,[170,240,333,1105,1329,1357,1777],[0.6931471805599453,0.6931471805599453,0.6931471805599453,0.0,0.28768207245178085,0.6931471805599453,0.6931471805599453])                       |
    * |0    |I wish Java could Spark use case classes |[i, wish, java, could, spark, use, case, classes]|(2000,[213,342,489,495,1105,1329,1809,1967],[1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0])|(2000,[213,342,489,495,1105,1329,1809,1967],[0.6931471805599453,0.6931471805599453,0.6931471805599453,0.6931471805599453,0.0,0.28768207245178085,0.6931471805599453,0.6931471805599453])|
    * |1    |Logistic regression Spark models are neat|[logistic, regression, spark, models, are, neat] |(2000,[286,695,1105,1138,1193,1604],[1.0,1.0,1.0,1.0,1.0,1.0])                |(2000,[286,695,1105,1138,1193,1604],[0.6931471805599453,0.6931471805599453,0.0,0.6931471805599453,0.6931471805599453,0.6931471805599453])                                               |
    * +-----+-----------------------------------------+-------------------------------------------------+------------------------------------------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
    * spark 在所有文档都出现过故值为0
    */

  rescaledData.select("features", "label").take(3).foreach(println)
  /**
    * [(2000,[170,240,333,1105,1329,1357,1777],[0.6931471805599453,0.6931471805599453,0.6931471805599453,0.0,0.28768207245178085,0.6931471805599453,0.6931471805599453]),0]
    * [(2000,[213,342,489,495,1105,1329,1809,1967],[0.6931471805599453,0.6931471805599453,0.6931471805599453,0.6931471805599453,0.0,0.28768207245178085,0.6931471805599453,0.6931471805599453]),0]
    * [(2000,[286,695,1105,1138,1193,1604],[0.6931471805599453,0.6931471805599453,0.0,0.6931471805599453,0.6931471805599453,0.6931471805599453]),1]
    */
}

3. Word2Vec

3.1 算法介绍

类别：estimator【评估器】

Word2Vec 是一种著名的 词嵌入（Word Embedding） 方法，它可以计算每个单词在其给定语料库环境下的 分布式词向量（Distributed Representation，亦直接被称为词向量）。
词向量表示可以在一定程度上刻画每个单词的语义。如果词的语义相近，它们的词向量在向量空间中也相互接近，这使得词语的向量化建模更加精确，可以改善现有方法并提高鲁棒性。
词向量已被证明在许多自然语言处理问题中具有非常重要的作用。如：机器翻译，标注问题，实体识别等
Word2Vec是一个Estimator，它采用一系列代表文档的词语来训练word2vecmodel。
该模型将每个词语映射到一个固定大小的向量。word2vecmodel使用文档中每个词语的平均数来将文档转换为向量，然后这个向量可以作为预测的特征，来计算文档相似度计算等等。
Word2Vec具有两种模型：

CBOW ：思想是通过每个词的上下文窗口词词向量来预测中心词的词向量。
Skip-gram：思想是通过每个中心词来预测其上下文窗口词，并根据预测结果来修正中心词的词向量。

在ml 库中，Word2Vec 的实现使用的是skip-gram模型。Skip-gram的训练目标是学习词表特征向量分布，其优化目标是在给定中心词的词向量情况下，最大化一下似然函数： $\frac{1}{T} \sum_{t=0}^1 \sum_{j=k-1}^{j=k} logp(W_{t+j}|W_t) $

其中 $w_1 … w_t$ 是一系列词序列，这里 $w_t$ 表示中心词，而 $w_{t+j}(j \in [-k,k])$ 是上下文窗口中的词。这里每个上下文窗口的词 $w_i$ 在给定中心词 $w_j$ 下的条件概率由 softmax 函数的形式计算，如下公式所示，其中 $u_w$ 和 $u_w$ 分别代表当前词的词向量以及当前上下文的词向量表示： $p(w_i|w_j) = \frac{exp(u_{w_i}^T u_{w_i})}{\sum_{l=1}^V exp(u_l^T u_{w_i})}$

因为 Skip-gram 模型使用的 softmax 计算较为复杂，所以 ml 与经典的额 Word2Vec 采用了相同的策略，使用 huffman 树来进行层次 softmax(hiercahical Softmax ) 方法来进行优化，使得 $log p (|w_i w_j)$ 计算复杂度从 $ O(V) $ 降到了 O(log(V))。

下面的代码段中，我们首先用一组文档，其中一个词语序列代表一个文档。对于每个文档，我们将其转换为一个特征向量。此特征向量可以被传递到一个学习算法。

3.2 代码示例


package hnbian.spark.ml.feature.extractors

import org.apache.spark.ml.feature.Word2Vec
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

/**
  * 特征提取 Word2Vec 测试代码
  * @author hnbian
  * @ Description
  * @ Date 2018/12/27 10:47
  **/
object Word2Vec extends App {
  val conf = new SparkConf().setAppName("Word2Vec")
  //设置master local[4] 指定本地模式开启模拟worker线程数
  conf.setMaster("local[4]")
  //创建sparkContext文件
  val sc = new SparkContext(conf)
  val spark = SparkSession.builder().getOrCreate()
  sc.setLogLevel("Error")

  //定义测试数据
  val documentDF = spark.createDataFrame(Seq(
    "Hi I heard about Spark".split(" "),
    "I wish Java could use case classes".split(" "),
    "Logistic regression models are neat".split(" ")
  ).map(Tuple1.apply)).toDF("text")

  documentDF.show(false)
  /**
    * +------------------------------------------+
    * |text                                      |
    * +------------------------------------------+
    * |[Hi, I, heard, about, Spark]              |
    * |[I, wish, Java, could, use, case, classes]|
    * |[Logistic, regression, models, are, neat] |
    * +------------------------------------------+
    */
  /**
    * 定义一个Word2Vec
    * 设置相应的超参数，这里设置特征向量的维度为3，
    * Word2Vec模型还有其他可设置的超参数，具体的超参数描述可以参见
    * http://spark.apache.org/docs/1.6.2/api/scala/index.html#org.apache.spark.ml.feature.Word2Vec
    */
  val word2Vec = new Word2Vec().
           setInputCol("text").
           setOutputCol("result").
           setVectorSize(5).
           setMinCount(0)

  //读入训练数据，调用fit()方法生成一个Word2VecModel
  val model = word2Vec.fit(documentDF)
  //使用Word2VecModel把文档转变成特征向量
  model.transform(documentDF).show(false)
  /**
    * +------------------------------------------+----------------------------------------------------------------+
    * |text                                      |result                                                          |
    * +------------------------------------------+----------------------------------------------------------------+
    * |[Hi, I, heard, about, Spark]              |[-0.008142343163490296,0.02051363289356232,0.03255096450448036] |
    * |[I, wish, Java, could, use, case, classes]|[0.043090314205203734,0.035048123182994974,0.023512658663094044]|
    * |[Logistic, regression, models, are, neat] |[0.038572299480438235,-0.03250147425569594,-0.01552378609776497]|
    * +------------------------------------------+----------------------------------------------------------------+
    */
  //文档被转变为了一个3维的特征向量，这些特征向量就可以被应用到相关的机器学习方法中。
}

4. CountVectorizer

4.1 算法介绍

类别：estimator【评估器】

CountVectorizer旨在通过计数来将一个文档转换为向量。当不存在先验字典时，CountVectorizer为Estimator取词汇进行训练，并生成一个 CountVectorizerModel 用于存储相应的词汇向量空间。该模型产生文档关于词语的稀疏表示，其表示可以传递给其他算法，例如LDA。

在CountVectorizerModel 的训练过程中，CountVectorizer将根据语料库中的词频排序从高到低进行选择，词汇表的最大含量由vocabsize超参数来指定，超参数minDF，则指定词汇表中的词语至少要在多少个不同文档中出现。

4.2 代码示例

package hnbian.spark.ml.feature.extractors

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.feature.{CountVectorizer, CountVectorizerModel}

/**
  * @author hnbian
  *         @ Description
  *         @ Date 2018/12/27 11:27
  **/
object CountVectorizer extends App {
  val conf = new SparkConf().setAppName("CountVectorizer")
  //设置master local[4] 指定本地模式开启模拟worker线程数
  conf.setMaster("local[4]")
  //创建sparkContext文件
  val sc = new SparkContext(conf)
  val spark = SparkSession.builder().getOrCreate()
  sc.setLogLevel("Error")
  val df = spark.createDataFrame(Seq(
    (0, Array("a", "b", "c", "E", "E", "E", "d", "d")),
    (1, Array("a", "b", "b", "E", "c", "a"))
  )).toDF("id", "words")

  df.show(false)
  /**
    * +---+------------------+
    * | id|             words|
    * +---+------------------+
    * |  0|      [a, b, c, d]|
    * |  1|[a, b, b, E, c, a]|
    * +---+------------------+
    *//**
    * 定义一个CountVectorizer，并设定相关超参数
    */
  val cvModel: CountVectorizerModel = new CountVectorizer()
    .setInputCol("words")
    .setOutputCol("features")
    .setVocabSize(3) //设定词汇表的最大量为3（最多选取三个词）
    .setMinDF(2) //设定词汇表中的词至少要在2个文档中出现过，以过滤那些偶然出现的词汇
    .fit(df)

  //可以通过CountVectorizerModel的vocabulary成员获得到模型的词汇表：
  println(cvModel.vocabulary.toBuffer)
  //ArrayBuffer(E, b, a)
  //d 虽然出现过两次但是只在一个文档中出现过
  //c 虽然满足设置的条件但是 出现次数只有两次 比其他词都少所以被过滤掉了

  //使用模型对文档进行转换
  cvModel.transform(df).show(false)
  /**
    * +---+------------------------+-------------------------+-------------------------------+
    * |id |words                   |features                 | 按照取词E、b、a的顺序求出现次数    |
    * +---+------------------------+-------------------------+-------------------------------+
    * |0  |[a, b, c, E, E, E, d, d]|(3,[0,1,2],[3.0,1.0,1.0])|  E 出现3次, b 1次, a 1次        |   
    * |1  |[a, b, b, E, c, a]      |(3,[0,1,2],[1.0,2.0,2.0])|  E 出现1次, b 2次, a 2次        | 
    * +---+------------------------+-------------------------+-------------------------------+
    */

  val cvm = new CountVectorizerModel(Array("a", "b", "c")).
    setInputCol("words").
    setOutputCol("features")

  println(cvm.vocabulary.toBuffer)
  //ArrayBuffer(a, b, c)
  cvm.transform(df).show(false)
  /**
    * +---+------------------------+-------------------------+--------------------------------+
    * |id |words                   |features                 | 按照取词a、b、c的顺序求出现次数     |
    * +---+------------------------+-------------------------+--------------------------------+
    * |0  |[a, b, c, E, E, E, d, d]|(3,[0,1,2],[1.0,1.0,1.0])| a 出现1次, b 1次, c 1次          |
    * |1  |[a, b, b, E, c, a]      |(3,[0,1,2],[2.0,2.0,1.0])| a 出现2次, b 2次, c 1次          |
    * +---+------------------------+-------------------------+--------------------------------+
    */
}