4.Spark特征提取、转换和选择 - 简书

标签: | 发表时间:2018-08-13 15:08 | 作者:
出处:https://www.jianshu.com

在实际机器学习项目中,我们获取的数据往往是不规范、不一致、有很多缺失数据,甚至不少错误数据,这些数据有时又称为脏数据或噪音,在模型训练前,务必对这些脏数据进行处理,否则,再好的模型,也只能脏数据进,脏数据出。

这章我们主要介绍对数据处理涉及的一些操作,主要包括:

  • 特征提取
  • 特征转换
  • 特征选择

4.1 特征提取

特征提取一般指从原始数据中抽取特征。

4.1.1 词频-逆向文件频率(TF-IDF)

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

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

      import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}  

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

val tokenizer = new Tokenizer().setInputCol("sentence").setOutputCol("words")  
val wordsData = tokenizer.transform(sentenceData)  
val hashingTF = new HashingTF().setInputCol("words").setOutputCol("rawFeatures").setNumFeatures(20)  
val featurizedData = hashingTF.transform(wordsData)  
// CountVectorizer也可获取词频向量  

val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")  
val idfModel = idf.fit(featurizedData)  
val rescaledData = idfModel.transform(featurizedData)  
rescaledData.select("features", "label").take(3).foreach(println)

4.1.2 Word2Vec

Word2vec是一个Estimator,它采用一系列代表文档的词语来训练word2vecmodel。 在下面的代码段中,我们首先用一组文档,其中每一个文档代表一个词语序列。对于每一个文档,我们将其转换为一个特征向量。此特征向量可以被传递到一个学习算法。

      import org.apache.spark.ml.feature.Word2Vec  

// 输入数据,每行为一个词袋,可来自语句或文档。
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")  

//训练从词到向量的映射  
val word2Vec = new Word2Vec()  
  .setInputCol("text")  
  .setOutputCol("result")  
  .setVectorSize(3)  
  .setMinCount(0)  
val model = word2Vec.fit(documentDF)  
val result = model.transform(documentDF)  
result.select("result").take(3).foreach(println)

4.1.3 计数向量器

计数向量器(Countvectorizer)和计数向量器模型(Countvectorizermodel)旨在通过计数来将一个文档转换为向量。

以下用实例来说明计数向量器的使用。
假设有以下列id和texts构成的DataFrame:

id texts
0 Array("a", "b", "c")
1 Array("a", "b", "b", "c", "a")

每行text都是Array [String]类型的文档。调用fit,CountVectorizer产生CountVectorizerModel含词汇(a,b,c)。转换后的输出列“向量”包含:
调用的CountVectorizer产生词汇(a,b,c)的CountVectorizerModel,转换后的输出向量如下:

id texts vector
0 Array("a", "b", "c") (3,[0,1,2],[1.0,1.0,1.0])
1 Array("a", "b", "b", "c", "a") (3,[0,1,2],[2.0,2.0,1.0])

每个向量代表文档的词汇表中每个词语出现的次数。

      import org.apache.spark.ml.feature.{CountVectorizer,CountVectorizerModel}

val df = spark.createDataFrame(Seq(
(0,Array("a","b","c")),
(1,Array("a","b","b","c","a"))
)).toDF("id","words")

//从语料库中拟合CountVectorizerModel
val cvModel:CountVectorizerModel=newCountVectorizer()
.setInputCol("words")
.setOutputCol("features")
.setVocabSize(3)
.setMinDF(2)
.fit(df)

//或者,用先验词汇表定义CountVectorizerModel
val cvm =newCountVectorizerModel(Array("a","b","c"))
.setInputCol("words")
.setOutputCol("features")

cvModel.transform(df).show(false)
+---+---------------+-------------------------+
|id |words          |features                 |
+---+---------------+-------------------------+
|0  |[a, b, c]      |(3,[0,1,2],[1.0,1.0,1.0])|
|1  |[a, b, b, c, a]|(3,[0,1,2],[2.0,2.0,1.0])|
+---+---------------+-------------------------+

4.2 特征转换

在机器学习中,数据处理是一件比较繁琐的事情,需要对原有特征做多种处理,如类型转换、标准化特征、新增衍生特征等等,需要耗费大量的时间和精力编写处理程序,不过,自从Spark推出ML后,情况大有改观,Spark ML包中提供了很多现成转换器,例如:StringIndexer、IndexToString、OneHotEncoder、VectorIndexer,它们提供了十分方便的特征转换功能,这些转换器类都位org.apache.spark.ml.feature包下。

4.2.1分词器

分词器(Tokenization)将文本划分为独立个体(通常为单词)。

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

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 =newTokenizer().setInputCol("sentence").setOutputCol("words")
val regexTokenizer =newRegexTokenizer()
.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)
+-----------------------------------+------------------------------------------+------+
|sentence                           |words                                     |tokens|
+-----------------------------------+------------------------------------------+------+
|Hi I heard about Spark             |[hi, i, heard, about, spark]              |5     |
|I wish Java could use case classes |[i, wish, java, could, use, case, classes]|7     |
|Logistic,regression,models,are,neat|[logistic,regression,models,are,neat]     |1     |
+-----------------------------------+------------------------------------------+------+
val regexTokenized = regexTokenizer.transform(sentenceDataFrame)
regexTokenized.select("sentence","words")
.withColumn("tokens", countTokens(col("words"))).show(false)
+-----------------------------------+------------------------------------------+------+
|sentence                           |words                                     |tokens|
+-----------------------------------+------------------------------------------+------+
|Hi I heard about Spark             |[hi, i, heard, about, spark]              |5     |
|I wish Java could use case classes |[i, wish, java, could, use, case, classes]|7     |
|Logistic,regression,models,are,neat|[logistic, regression, models, are, neat] |5     |
+-----------------------------------+------------------------------------------+------+

4.2.2 移除停用词

停用词为在文档中频繁出现,但未承载太多意义的词语,它们不应该被包含在算法输入中,所以会用到移除停用词(StopWordsRemover)。

示例:
假设我们有如下DataFrame,有id和raw两列

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

通过对raw列调用StopWordsRemover,我们可以得到筛选出的结果列如下

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

其中,“I”, “the”, “had”以及“a”被移除。
实现以上功能的详细代码:

      import org.apache.spark.ml.feature.StopWordsRemover

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)

4.2.3 n-gram

一个n-gram是一个长度为整数n的字序列。NGram可以用来将输入转换为n-gram。

      import org.apache.spark.ml.feature.NGram
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")

val ngram = new NGram().setN(2).setInputCol("words").setOutputCol("ngrams")

val ngramDataFrame = ngram.transform(wordDataFrame)
ngramDataFrame.select("ngrams").show(false)
+------------------------------------------------------------------+
|ngrams                                                            |
+------------------------------------------------------------------+
|[Hi I, I heard, heard about, about Spark]                         |
|[I wish, wish Java, Java could, could use, use case, case classes]|
|[Logistic regression, regression models, models are, are neat]    |
+------------------------------------------------------------------+

4.2.4 二值化

二值化,通过设置阀值,将连续型的特征转化为两个值。大于阀值为1,否则为0。
注:以下规范化操作一般是针对一个特征向量(dataFrame中的一个colum)来操作的。

      import org.apache.spark.ml.feature.Binarizer

val data = Array((0, 0.1), (1, 0.8), (2, 0.2))
val dataFrame = spark.createDataFrame(data).toDF("id", "feature")
+---+-------+
| id|feature|
+---+-------+
|  0|    0.1|
|  1|    0.8|
|  2|    0.2|
+---+-------+
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()
+---+-------+-----------------+
| id|feature|binarized_feature|
+---+-------+-----------------+
|  0|    0.1|              0.0|
|  1|    0.8|              1.0|
|  2|    0.2|              0.0|
+---+-------+-----------------+

4.2.5 主成分分析

主成分分析被广泛应用在各种统计学、机器学习问题中,是最常见的降维方法之一。
PCA在Spark2.0用法比较简单,只需要设置:

      .setInputCol(“features”)//保证输入是特征值向量
.setOutputCol(“pcaFeatures”)//输出
.setK(3)//主成分个数
注意:PCA前一定要对特征向量进行规范化(标准化)!!!
      import org.apache.spark.ml.feature.PCA
import org.apache.spark.ml.feature.PCAModel//不是mllib
import org.apache.spark.ml.feature.StandardScaler
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession

val rawDataFrame=spark.read.format("libsvm").load("file:///home/hadoop/bigdata/spark/data/mllib/sample_libsvm_data.txt")

val scaledDataFrame=new StandardScaler()
                  .setInputCol("features")
                  .setOutputCol("scaledFeatures")
                  .setWithMean(false)//对于稀疏数据(如本次使用的数据),不要使用平均值
                  .setWithStd(true)
                  .fit(rawDataFrame)
                  .transform(rawDataFrame)
//PCA Model
val pcaModel=new PCA().setInputCol("scaledFeatures")
                      .setOutputCol("pcaFeatures")
                      .setK(3)//
                      .fit(scaledDataFrame)
//进行PCA降维
pcaModel.transform(scaledDataFrame).select("label","pcaFeatures").show(10,false)
//没有标准化特征向量,直接进行PCA主成分:各主成分之间值变化太大,有数量级的差别。
//标准化特征向量后PCA主成分,各主成分之间值基本上在同一水平上,结果更合理
+-----+-------------------------------------------------------------+
|label|pcaFeatures                                                  |
+-----+-------------------------------------------------------------+
|0.0  |[-14.998868464839624,-10.137788261664621,-3.042873539670117] |
|1.0  |[2.1965800525589754,-4.139257418439533,-11.386135042845101]  |
|1.0  |[1.0254645688925883,-0.8905813756164163,7.168759904518129]   |
|1.0  |[1.5069317554093433,-0.7289177578028571,5.23152743564543]    |
|1.0  |[1.6938250375084654,-0.4350617717494331,4.770263568537382]   |
|0.0  |[-15.870371979062549,-9.999445137658528,-6.521920373215663]  
//如何选择k值?
val  pcaModel=new PCA().setInputCol("scaledFeatures")
                      .setOutputCol("pcaFeatures")
                      .setK(50)//
                      .fit(scaledDataFrame)
var i=1
for( x<-pcaModel.explainedVariance.toArray){
        println(i+"\t"+x+"  ")
        i +=1
        }
//运行结果(前10行),随着k的增加,精度趋于平稳。
1   0.25934799275530857  
2   0.12355355301486977  
3   0.07447670060988294  
4   0.0554545717486928  
5   0.04207050513264405  
6   0.03715986573644129  
7   0.031350566055423544  
8   0.027797304129489515  
9   0.023825873477496748  
10  0.02268054946233242

4.2.6 多项式展开

多项式展开(PolynomialExpansion)即通过产生n维组合将原始特征将特征扩展到多项式空间。下面的示例会介绍如何将你的特征集拓展到3维多项式空间。

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

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)
+----------+------------------------------------------+
|features  |polyFeatures                              |
+----------+------------------------------------------+
|[2.0,1.0] |[2.0,4.0,8.0,1.0,2.0,4.0,1.0,2.0,1.0]     |
|[0.0,0.0] |[0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0]     |
|[3.0,-1.0]|[3.0,9.0,27.0,-1.0,-3.0,-9.0,1.0,3.0,-1.0]|
+----------+------------------------------------------+

4.2.7 离散余弦变换

离散余弦变换(DCT)是与傅里叶变换相关的一种变换,它类似于离散傅立叶变换,但是只使用实数。

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

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)
+----------------------------------------------------------------+
|featuresDCT                                                     |
+----------------------------------------------------------------+
|[1.0,-1.1480502970952693,2.0000000000000004,-2.7716385975338604]|
|[-1.0,3.378492794482933,-7.000000000000001,2.9301512653149677]  |
|[4.0,9.304453421915744,11.000000000000002,1.5579302036357163]   |
+----------------------------------------------------------------+

4.2.8 字符串-索引变换

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

字符串—索引变换(StringIndexer)是将字符串列编码为标签索引列。示例数据为一个含有id和category两列的DataFrame

id category
0 a
1 b
2 c
3 a
4 a
5 c
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

category是有3种取值的字符串列(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
id category
0 a
1 b
2 c
3 d

a获得索引0,因为它是最频繁的,随后是具有索引1的c和具有索引2的b。
如果测试数据集中比训练数据集多了一个d类:

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

如果您没有设置StringIndexer如何处理未看见的标签(默认值)或将其设置为“错误”,则会抛出异常。 但是,如果您调用了setHandleInvalid(“skip”),d类将不出现,结果为以下数据集:

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

以下是使用StringIndexer的一个示例:

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

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")

val indexed = indexer.fit(df).transform(df)
indexed.show()
+---+--------+-------------+                                                    
| 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|
+---+--------+-------------+

4.2.9 索引-字符串变换

id categoryIndex
0 0.0
1 2.0
2 1.0
3 0.0
4 0.0
5 1.0

与StringIndexer对应,索引—字符串变换(IndexToString)是将指标标签映射回原始字符串标签。

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

应用IndexToString,将categoryIndex作为输入列,将originalCategory作为输出列,我们可以检索我们的原始标签(它们将从列的元数据中推断):

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

以下是以上整个过程的一个实例:

      import org.apache.spark.ml.attribute.Attribute
import org.apache.spark.ml.feature.{IndexToString, StringIndexer}

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")
  .fit(df)
val indexed = indexer.transform(df)

println(s"Transformed string column '${indexer.getInputCol}' " +
    s"to indexed column '${indexer.getOutputCol}'")
indexed.show()

val inputColSchema = indexed.schema(indexer.getOutputCol)
println(s"StringIndexer will store labels in output column metadata: " +
    s"${Attribute.fromStructField(inputColSchema).toString}\n")

val converter = new IndexToString()
  .setInputCol("categoryIndex")
  .setOutputCol("originalCategory")

val converted = converter.transform(indexed)

println(s"Transformed indexed column '${converter.getInputCol}' back to original string " +
    s"column '${converter.getOutputCol}' using labels in metadata")
converted.select("id", "categoryIndex", "originalCategory").show()

4.2.10 独热编码

独热编码(OneHotEncoder)将标签指标映射为二值向量,其中最多一个单值。

      import org.apache.spark.ml.feature.{OneHotEncoder, StringIndexer}

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")
  .fit(df)
val indexed = indexer.transform(df)

val encoder = new OneHotEncoder()
  .setInputCol("categoryIndex")
  .setOutputCol("categoryVec")
.setDropLast(false)

val encoded = encoder.transform(indexed)
encoded.show()

【说明】
1、OneHotEncoder缺省状态下将删除最后一个分类或把最后一个分类作为0.
//示例

      import org.apache.spark.ml.feature.{OneHotEncoder, StringIndexer}

val fd = spark.createDataFrame( Seq((1.0, "a"), (1.5, "a"), (10.0, "b"), (3.2, "c"),(3.8,"c"))).toDF("x","c")
val ss =new StringIndexer().setInputCol("c").setOutputCol("c_idx")
val ff = ss.fit(fd).transform(fd)
ff.show()

显示结果如下:

      +----+---+-----+
| x| c|c_idx|
+----+---+-----+
| 1.0| a| 0.0|
| 1.5| a| 0.0|
|10.0| b| 2.0|
| 3.2| c| 1.0|
| 3.8| c| 1.0|
+----+---+-----+

最后一个分类为b,通过OneHotEncoder变为向量后,已被删除。

      val oe = new OneHotEncoder().setInputCol("c_idx").setOutputCol("c_idx_vec")
val fe = oe.transform(ff)
fe.show()

显示结果如下:

      +----+---+-----+-------------+
| x| c|c_idx| c_idx_vec|
+----+---+-----+-------------+
| 1.0| a| 0.0|(2,[0],[1.0])|
| 1.5| a| 0.0|(2,[0],[1.0])|
|10.0| b| 2.0| (2,[],[])|
| 3.2| c| 1.0|(2,[1],[1.0])|
| 3.8| c| 1.0|(2,[1],[1.0])|
+----+---+-----+-------------+

与其他特征组合为特征向量后,将置为0,请看下例

      val assembler = new VectorAssembler().setInputCols(Array("x", "c_idx", "c_idx_vec")).setOutputCol("features")
val vecDF: DataFrame = assembler.transform(fe)
vecDF.show(false)

显示结果如下:

      +----+---+-----+-------------+------------------+
|x |c |c_idx|c_idx_vec |features |
+----+---+-----+-------------+------------------+
|1.0 |a |0.0 |(2,[0],[1.0])|[1.0,0.0,1.0,0.0] |
|1.5 |a |0.0 |(2,[0],[1.0])|[1.5,0.0,1.0,0.0] |
|10.0|b |2.0 |(2,[],[]) |[10.0,2.0,0.0,0.0]|
|3.2 |c |1.0 |(2,[1],[1.0])|[3.2,1.0,0.0,1.0] |
|3.8 |c |1.0 |(2,[1],[1.0])|[3.8,1.0,0.0,1.0] |
+----+---+-----+-------------+------------------+

如果想不删除最后一个分类,可添加setDropLast(False)。

      oe.setDropLast(false)
val fl = oe.transform(ff)
fl.show()

显示结果如下:

      +----+---+-----+-------------+
| x| c|c_idx| c_idx_vec|
+----+---+-----+-------------+
| 1.0| a| 0.0|(3,[0],[1.0])|
| 1.5| a| 0.0|(3,[0],[1.0])|
|10.0| b| 2.0|(3,[2],[1.0])|
| 3.2| c| 1.0|(3,[1],[1.0])|
| 3.8| c| 1.0|(3,[1],[1.0])|
+----+---+-----+-------------+

与其他特征向量结合后,情况如下:

      val vecDFl: DataFrame = assembler.transform(fl)
vecDFl.show(false)

显示结果如下:

      +----+---+-----+-------------+----------------------+
|x |c |c_idx|c_idx_vec |features |
+----+---+-----+-------------+----------------------+
|1.0 |a |0.0 |(3,[0],[1.0])|(5,[0,2],[1.0,1.0]) |
|1.5 |a |0.0 |(3,[0],[1.0])|(5,[0,2],[1.5,1.0]) |
|10.0|b |2.0 |(3,[2],[1.0])|[10.0,2.0,0.0,0.0,1.0]|
|3.2 |c |1.0 |(3,[1],[1.0])|[3.2,1.0,0.0,1.0,0.0] |
|3.8 |c |1.0 |(3,[1],[1.0])|[3.8,1.0,0.0,1.0,0.0] |
+----+---+-----+-------------+----------------------+

2、如果分类中出现空字符,需要进行处理,如设置为"None",否则会报错。

4.2.11 向量-索引变换

在下面的例子中,我们读取一个标记点的数据集,然后使用VectorIndexer来决定哪些特征应该被视为分类。我们将分类特征值转换为它们的索引。这个变换的数据然后可以被传递到诸如DecisionTreeRegressor的处理分类特征的算法。

      import org.apache.spark.ml.feature.VectorIndexer

val data = spark.read.format("libsvm").load("file:///home/hadoop/bigdata/spark/data/mllib/sample_libsvm_data.txt")

val indexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexed")
  .setMaxCategories(10)

val indexerModel = indexer.fit(data)

val categoricalFeatures: Set[Int] = indexerModel.categoryMaps.keys.toSet
println(s"Chose ${categoricalFeatures.size} categorical features: " +
  categoricalFeatures.mkString(", "))

// Create new column "indexed" with categorical values transformed to indices
val indexedData = indexerModel.transform(data)
indexedData.show()
+-----+--------------------+--------------------+
|label|            features|             indexed|
+-----+--------------------+--------------------+
|  0.0|(692,[127,128,129...|(692,[127,128,129...|
|  1.0|(692,[158,159,160...|(692,[158,159,160...|
|  1.0|(692,[124,125,126...|(692,[124,125,126...|
|  1.0|(692,[152,153,154...|(692,[152,153,154...|
|  1.0|(692,[151,152,153...|(692,[151,152,153...|
|  0.0|(692,[129,130,131...|(692,[129,130,131...|

4.2.12交互式

例子,假设我们有以下DataFrame的列“id1”,“vec1”和“vec2”

id1 vec1 vec2
1 [1.0,2.0,3.0] [8.0,4.0,5.0]
2 [4.0,3.0,8.0] [7.0,9.0,8.0]
3 [6.0,1.0,9.0] [2.0,3.0,6.0]
4 [10.0,8.0,6.0] [9.0,4.0,5.0]
5 [9.0,2.0,7.0] [10.0,7.0,3.0]
6 [1.0,1.0,4.0] [2.0,8.0,4.0]

应用与这些输入列的交互,然后interactionedCol作为输出列包含:

id1 vec1 vec2 interactedCol
1 [1.0,2.0,3.0] [8.0,4.0,5.0] [8.0,4.0,5.0,16.0,8.0,10.0,24.0,12.0,15.0]
2 [4.0,3.0,8.0] [7.0,9.0,8.0] [56.0,72.0,64.0,42.0,54.0,48.0,112.0,144.0,128.0]
3 [6.0,1.0,9.0] [2.0,3.0,6.0] [36.0,54.0,108.0,6.0,9.0,18.0,54.0,81.0,162.0]
4 [10.0,8.0,6.0] [9.0,4.0,5.0] [360.0,160.0,200.0,288.0,128.0,160.0,216.0,96.0,120.0]
5 [9.0,2.0,7.0] [10.0,7.0,3.0] [450.0,315.0,135.0,100.0,70.0,30.0,350.0,245.0,105.0]
6 [1.0,1.0,4.0] [2.0,8.0,4.0] [12.0,48.0,24.0,12.0,48.0,24.0,48.0,192.0,96.0]

以下是实现以上转换的具体代码:

      import org.apache.spark.ml.feature.Interaction
import org.apache.spark.ml.feature.VectorAssembler

val df = spark.createDataFrame(Seq(
  (1, 1, 2, 3, 8, 4, 5),
  (2, 4, 3, 8, 7, 9, 8),
  (3, 6, 1, 9, 2, 3, 6),
  (4, 10, 8, 6, 9, 4, 5),
  (5, 9, 2, 7, 10, 7, 3),
  (6, 1, 1, 4, 2, 8, 4)
)).toDF("id1", "id2", "id3", "id4", "id5", "id6", "id7")

val assembler1 = new VectorAssembler().
  setInputCols(Array("id2", "id3", "id4")).
  setOutputCol("vec1")

val assembled1 = assembler1.transform(df)

val assembler2 = new VectorAssembler().
  setInputCols(Array("id5", "id6", "id7")).
  setOutputCol("vec2")

val assembled2 = assembler2.transform(assembled1).select("id1", "vec1", "vec2")

val interaction = new Interaction()
  .setInputCols(Array("id1", "vec1", "vec2"))
  .setOutputCol("interactedCol")

val interacted = interaction.transform(assembled2)

interacted.show(truncate = false)

4.2.13正则化

以下示例演示如何加载libsvm格式的数据集,然后将每行标准化为具有单位L1范数和单位L∞范数。

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

val dataFrame = spark.createDataFrame(Seq(
  (0, Vectors.dense(1.0, 0.5, -1.0)),
  (1, Vectors.dense(2.0, 1.0, 1.0)),
  (2, Vectors.dense(4.0, 10.0, 2.0))
)).toDF("id", "features")

// 使用L^1正规化向量
val normalizer = new Normalizer()
  .setInputCol("features")
  .setOutputCol("normFeatures")
  .setP(1.0)

val l1NormData = normalizer.transform(dataFrame)
println("Normalized using L^1 norm")
l1NormData.show()

//使用L^∞正规化向量..
val lInfNormData = normalizer.transform(dataFrame, normalizer.p -> Double.PositiveInfinity)
println("Normalized using L^inf norm")
lInfNormData.show()

4.2.14规范化(StandardScaler)

以下示例演示如何以libsvm格式加载数据集,然后规范化每个要素的单位标准偏差。

      import org.apache.spark.ml.feature.StandardScaler

val dataFrame = spark.read.format("libsvm").load("file:///u01/bigdata/spark/data/mllib/sample_libsvm_data.txt")

val scaler = new StandardScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")
  .setWithStd(true)
  .setWithMean(false)

//通过拟合StandardScaler来计算汇总统计
val scalerModel = scaler.fit(dataFrame)

//标准化每个特征使其有单位标准偏差
val scaledData = scalerModel.transform(dataFrame)
scaledData.show()

4.2.15最大值-最小值缩放

下面的示例展示如果读入一个libsvm形式的数据以及调整其特征值到[0,1]之间。
调用示例:

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

val dataFrame = spark.createDataFrame(Seq(
  (0, Vectors.dense(1.0, 0.1, -1.0)),
  (1, Vectors.dense(2.0, 1.1, 1.0)),
  (2, Vectors.dense(3.0, 10.1, 3.0))
)).toDF("id", "features")

val scaler = new MinMaxScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")

//进行汇总统计并生成MinMaxScalerModel
val scalerModel = scaler.fit(dataFrame)

//将每个特征重新缩放至[min,max]范围
val scaledData = scalerModel.transform(dataFrame)
println(s"Features scaled to range: [${scaler.getMin}, ${scaler.getMax}]")
scaledData.select("features", "scaledFeatures").show()

显示结果如下:

      Features scaled to range: [0.0, 1.0]
+--------------+--------------+
| features|scaledFeatures|
+--------------+--------------+
|[1.0,0.1,-1.0]| [0.0,0.0,0.0]|
| [2.0,1.1,1.0]| [0.5,0.1,0.5]|
|[3.0,10.1,3.0]| [1.0,1.0,1.0]|
+--------------+--------------+

4.2.16最大值-绝对值缩放

以下示例演示如何加载libsvm格式的数据集,然后将每个特征重新缩放到[-1,1]。

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

val dataFrame = spark.createDataFrame(Seq(
  (0, Vectors.dense(1.0, 0.1, -8.0)),
  (1, Vectors.dense(2.0, 1.0, -4.0)),
  (2, Vectors.dense(4.0, 10.0, 8.0))
)).toDF("id", "features")

val scaler = new MaxAbsScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")

// 进行汇总统计并生成MaxAbsScalerModel
val scalerModel = scaler.fit(dataFrame)

// rescale each feature to range [-1, 1]
val scaledData = scalerModel.transform(dataFrame)
scaledData.select("features", "scaledFeatures").show()

运行结果如下:

      +--------------+----------------+
| features| scaledFeatures|
+--------------+----------------+
|[1.0,0.1,-8.0]|[0.25,0.01,-1.0]|
|[2.0,1.0,-4.0]| [0.5,0.1,-0.5]|
|[4.0,10.0,8.0]| [1.0,1.0,1.0]|
+--------------+----------------+

4.2.17离散化重组

以下示例演示如何将双列列存储到另一个索引列的列中。

      import org.apache.spark.ml.feature.Bucketizer

val splits = Array(Double.NegativeInfinity, -0.5, 0.0, 0.5, Double.PositiveInfinity)

val data = Array(-999.9, -0.5, -0.3, 0.0, 0.2, 999.9)
val dataFrame = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")

val bucketizer = new Bucketizer()
  .setInputCol("features")
  .setOutputCol("bucketedFeatures")
  .setSplits(splits)

// 把原来的数据转换为箱式索引
val bucketedData = bucketizer.transform(dataFrame)

println(s"Bucketizer output with ${bucketizer.getSplits.length-1} buckets")
bucketedData.show()

运行结果如下:

      +--------+----------------+
|features|bucketedFeatures|
+--------+----------------+
| -999.9| 0.0|
| -0.5| 1.0|
| -0.3| 1.0|
| 0.0| 2.0|
| 0.2| 2.0|
| 999.9| 3.0|
+--------+----------------+

4.2.18元素乘积

下面的示例演示了如何使用变换向量值来变换向量

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

//创建一些向量数据; 也适用于稀疏向量。 
val dataFrame = spark.createDataFrame(Seq(
  ("a", Vectors.dense(1.0, 2.0, 3.0)),
  ("b", Vectors.dense(4.0, 5.0, 6.0)))).toDF("id", "vector")

val transformingVector = Vectors.dense(0.0, 1.0, 2.0)
val transformer = new ElementwiseProduct()
  .setScalingVec(transformingVector)
  .setInputCol("vector")
  .setOutputCol("transformedVector")

//创建新列,可批量转换向量。
transformer.transform(dataFrame).show()

运行结果如下:

      +---+-------------+-----------------+
| id| vector|transformedVector|
+---+-------------+-----------------+
| a|[1.0,2.0,3.0]| [0.0,2.0,6.0]|
| b|[4.0,5.0,6.0]| [0.0,5.0,12.0]|
+---+-------------+-----------------+

4.2.19 SQL转换器

假设我们有以下DataFrame和列id,v1和v2

id v1 v2
0 1.0 3.0
2 2.0 5.0

这是SQLTransformer "SELECT *, (v1 + v2) AS v3, (v1 * v2) AS v4 FROM THIS":语句的输出。

id v1 v2 v3 v4
0 1.0 3.0 4.0 3.0
2 2.0 5.0 7.0 10.0

以下是实现以上结果的具体代码:

      import org.apache.spark.ml.feature.SQLTransformer

val df = spark.createDataFrame(
  Seq((0, 1.0, 3.0), (2, 2.0, 5.0))).toDF("id", "v1", "v2")

val sqlTrans = new SQLTransformer().setStatement(
  "SELECT *, (v1 + v2) AS v3, (v1 * v2) AS v4 FROM __THIS__")

sqlTrans.transform(df).show()

4.2.20向量汇编

例子
假设我们有一个带有id,hour,mobile,userFeatures和clicked列的DataFrame:

id hour mobile userFeatures clicked
0 18 1.0 [0.0, 10.0, 0.5] 1.0

userFeatures是一个包含三个用户特征的向量列。我们希望将hour,mobile和userFeatures合并成一个称为特征的单一特征向量,并使用它来预测是否被点击。如果我们将VectorAssembler的输入列设置为hour,mobile和userFeatures,并将列输出到特征,则在转换后,我们应该得到以下DataFrame:

id hour mobile userFeatures clicked features
0 18 1.0 [0.0, 10.0, 0.5] 1.0 [18.0, 1.0, 0.0, 10.0, 0.5]

以下是实现上述功能的代码:

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

val dataset = spark.createDataFrame(
  Seq((0, 18, 1.0, Vectors.dense(0.0, 10.0, 0.5), 1.0))
).toDF("id", "hour", "mobile", "userFeatures", "clicked")

val assembler = new VectorAssembler()
  .setInputCols(Array("hour", "mobile", "userFeatures"))
  .setOutputCol("features")

val output = assembler.transform(dataset)
println("Assembled columns 'hour', 'mobile', 'userFeatures' to vector column 'features'")
output.select("features", "clicked").show(false)

4.2.21分位数离散化

示例:
假设我们有如下DataFrame包含id,hour:

id hour
0 18.0
---- ------
1 19.0
---- ------
2 8.0
---- ------
3 5.0
---- ------
4 2.2

hour是Double类型的连续特征。我们希望将连续特征变成一个分级特征。给定numBuckets = 3,我们可得到以下DataFrame:

id hour result
0 18.0 2.0
---- ------ ------
1 19.0 2.0
---- ------ ------
2 8.0 1.0
---- ------ ------
3 5.0 1.0
---- ------ ------
4 2.2 0.0

实现以上功能的scala代码如下:

      import org.apache.spark.ml.feature.QuantileDiscretizer

val data = Array((0, 18.0), (1, 19.0), (2, 8.0), (3, 5.0), (4, 2.2))
val df = spark.createDataFrame(data).toDF("id", "hour")

val discretizer = new QuantileDiscretizer()
  .setInputCol("hour")
  .setOutputCol("result")
  .setNumBuckets(3)

val result = discretizer.fit(df).transform(df)
result.show()

4.3 特征选择

特征选择(Feature Selection)是从特征向量中选择那些更有效的特征,组成新的、更简单有效的特征向量的过程。它在数据分析中常用使用,尤其在高维数据分析中,可以剔除冗余或影响不大的特征,提升模型的性能。

4.3.1 向量机

假设我们有一个DataFrame与列userFeatures:

userFeatures

[0.0, 10.0, 0.5]

userFeatures是一个包含三个用户特征的向量列。假设userFeature的第一列全部为零,因此我们要删除它并仅选择最后两列。 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"]

以下是实现向量选择的一个scala代码示例

      import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType

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)

运行结果:

      +--------------------+-------------+
|userFeatures |features |
+--------------------+-------------+
|(3,[0,1],[-2.0,2.3])|(2,[0],[2.3])|
|[-2.0,2.3,0.0] |[2.3,0.0] |
+--------------------+-------------+

4.3.2 R公式

示例:
假设我们有一个DataFrame含有id,country, hour和clicked四列:

id country hour clicked
7 "US" 18 1.0
8 "CA" 12 0.0
9 "NZ" 15 0.0

如果我们使用RFormula公式clicked ~ country+ hour,则表明我们希望基于country和hour预测clicked,通过转换我们可以得到如下DataFrame:

id country hour clicked features label
7 "US" 18 1.0 [0.0, 0.0, 18.0] 1.0
8 "CA" 12 0.0 [0.0, 1.0, 12.0] 0.0
9 "NZ" 15 0.0 [1.0, 0.0, 15.0] 0.0

以下是实现上述功能的scala代码:

      import org.apache.spark.ml.feature.RFormula

val dataset = spark.createDataFrame(Seq(
  (7, "US", 18, 1.0),
  (8, "CA", 12, 0.0),
  (9, "NZ", 15, 0.0)
)).toDF("id", "country", "hour", "clicked")

val formula = new RFormula()
  .setFormula("clicked ~ country + hour")
  .setFeaturesCol("features")
  .setLabelCol("label")

val output = formula.fit(dataset).transform(dataset)
output.select("features", "label").show()

4.3.3 卡方特征选择

示例:
假设我们有一个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,features中最后一列将会是最有用特征:

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]

使用ChiSqSelector的scala代码示例:

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

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

结果显示:

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

4.4 小结

本章主要介绍了对数据特征或变量的一些常用操作,包括特征提取,特征转换以及特征选择等方法,这些任务在实际项目中往往花费大量时间和精力,尤其要自己编写这方面的代码或函数,更是如此,Spark ML目前提供了很多现成函数,有效使用这些函数将有助于提供我们开发效率,同时使我们有更多时间优化或提升模型性能。下一章我们将介绍优化或提升模型性能一些方法。

相关 [spark 特征提取 选择] 推荐:

4.Spark特征提取、转换和选择 - 简书

- -
在实际机器学习项目中,我们获取的数据往往是不规范、不一致、有很多缺失数据,甚至不少错误数据,这些数据有时又称为脏数据或噪音,在模型训练前,务必对这些脏数据进行处理,否则,再好的模型,也只能脏数据进,脏数据出. 这章我们主要介绍对数据处理涉及的一些操作,主要包括:. 特征提取一般指从原始数据中抽取特征.

Spark-mllib 文本特征提取算法 - CSDN博客

- -
Spark MLlib 提供三种文本特征提取方法,分别为TF-IDF、Word2Vec以及CountVectorizer,. 词频-逆向文件频率(TF-IDF)是一种在文本挖掘中广泛使用的特征向量化方法,它可以体现一个文档中词语在语料库中的重要程度. 词语由t表示,文档由d表示,语料库由D表示. 词频TF(t,,d)是词语t在文档d中出现的次数.

scikit-learn中的文本特征提取

- - 标点符
文本分析是机器学习算法的主要应用领域. 由于大部分机器学习算法只能接收固定长度的数值型矩阵特征,导致文本字符串等并不能直接被使用,针对此问题scikit-learn提供了将文本转化为数值型特征的方法,今天就一起来学习下. scikit-learn中的sklearn.feature_extraction.text提供了将文本转化为特征向量的工具:.

【图像分析】低层次特征提取

- - CSDN博客推荐文章
本节主要介绍提取低层次特征信息的方要方法. 在有些情形下,这些信息对理解图是足够的. 但是,低层次特征提取函数的目的通常是为更高级分析提供信息. 用于边界,角点和运动提取的方法,非常具有Robustness以及性能特久所必备的特性. 低层次特征是不需要任何形状/空间关系的信息就可以从图像中自动提取的基本特征,如:常用地阈值方法就是作为点处理的一种低层次特征提方式.

浅谈模式识别中的特征提取

- - CSDN博客综合推荐文章
      这两天一直在看深度学习的东西,看的头晕脑胀,不过晕乎归晕乎,感觉对模式识别中的特征提取有了更深一点的小理解,暂时记载下来.   突然觉得,模式识别的所有问题都绕不过两个关键门槛,第一是分类器,第二便是特征提取. 而且几乎所有模式识别方面的研究都是在优化这两个问题,要么是造一个更牛的分类器,要么是找出一些表现力更高的特征形式.

图像处理之特征提取 - 简书

- -
目前火热的 Deep Learning 会灭绝传统的 SIFT / SURF 特征提取方法吗. 由于之前研究过SIFT和HOG这两种传统的特征提取方法,故本篇文章先对SIFT和HOG作一综述,并比较二者优缺点. 之后,将SIFT和HOG同神经网络特征提取做一对比,浅谈对上述问题的看法. 如果能写得快一些,再简单介绍其他几种传统的特征提取的方法——SURF、ORB、LBP、HAAR等等.

Spark概览

- - 简单文本
Spark具有先进的DAG执行引擎,支持cyclic data flow和内存计算. 因此,它的运行速度,在内存中是Hadoop MapReduce的100倍,在磁盘中是10倍. 这样的性能指标,真的让人心动啊. Spark的API更为简单,提供了80个High Level的操作,可以很好地支持并行应用.

Spark与Mapreduce?

- - 崔永键的博客
我本人是类似Hive平台的系统工程师,我对MapReduce的熟悉程度是一般,它是我的底层框架. 我隔壁组在实验Spark,想将一部分计算迁移到Spark上. 年初的时候,看Spark的评价,几乎一致表示,Spark是小数据集上处理复杂迭代的交互系统,并不擅长大数据集,也没有稳定性. 但是最近的风评已经变化,尤其是14年10月他们完成了Peta sort的实验,这标志着Spark越来越接近替代Hadoop MapReduce了.

Spark迷思

- - ITeye博客
目前在媒体上有很大的关于Apache Spark框架的声音,渐渐的它成为了大数据领域的下一个大的东西. 证明这件事的最简单的方式就是看google的趋势图:. 上图展示的过去两年Hadoop和Spark的趋势. Spark在终端用户之间变得越来越受欢迎,而且这些用户经常在网上找Spark相关资料. 这给了Spark起了很大的宣传作用;同时围绕着它的也有误区和思维错误,而且很多人还把这些误区作为银弹,认为它可以解决他们的问题并提供比Hadoop好100倍的性能.

Spark 优化

- - CSDN博客推荐文章
提到Spark与Hadoop的区别,基本最常说的就是Spark采用基于内存的计算方式,尽管这种方式对数据处理的效率很高,但也会往往引发各种各样的问题,Spark中常见的OOM等等. 效率高的特点,注定了Spark对性能的严苛要求,那Spark不同程序的性能会碰到不同的资源瓶颈,比如:CPU,带宽、内存.