Spark ML的特征处理实战

本文来自OPPO互联网技术团队，转载请注名做者。同时欢迎关注咱们的公众号：OPPO_tech，与你分享OPPO前沿互联网技术及活动。

一 、特征处理的意义

一般状况下，咱们获得的数据中包含脏数据或者噪声。在模型训练前，须要对这些数据进行预处理，不然再好的模型也只能“garbage in，garbage out”。

数据预处理主要包括三部分，特征提取、特征转换和特征选择。

2、特征提取

特征提取通常指从原始数据中抽取特征的过程。

1. 计数向量器(Countvectorizer)

(1) 定义及用途:计数向量器将全部的文本词语进行编号，并统计该词语在文档中的词频做为特征向量。

(2) spark ml中代码实例

import org.apache.spark.ml.feature.{CountVectorizer, CountVectorizerModel}
val df = spark.createDataFrame(Seq(
      (0, Array("a","e","a","d","b")),
      (1, Array("a","c","b","d","c","f","a","b")),
      (2, Array("a","f"))
)).toDF("id", "words")    
var cv_model = new CountVectorizer().setInputCol("words").setOutputCol("features").setVocabSize(10).setMinDF(2).fit(df)
val cv1 = cv_model.transform(df)
cv1.show(false)

注意点：计数向量器会将全部数据整合到一块儿进行去重造成一张词表，经过 setVocabSize 和 setMinDF 这两个参数来肯定是否进入词表。其中 setVocabSize 决定词表的长度，而 setMinDF 决定要在多少个不一样的样本中出现才进入词表。上例中设置词表的长度为10，至少在两个样本中出现才会进入词表，则能进入词表的只有a,b,d,f。c和e仅在一条数据中出现因此并不会统计词频。

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

(1) 定义及用途：通俗的理解就是计算一个词区别一篇文档的程度。经过在一篇文档中的词频和该词在文档库中多少篇文档中出现综合来评估。仅经过词频来区分一篇文档是不合理的。

好比文档中会屡次出现能表明通用含义的词，可是这些词对于文档的识别并没有意义。咱们须要的是一些特别的词，它出现的次数多，而且能在少数的文档中出现，这些词才可以识别文档。举个极端的例子，好比“咱们”这个词可能出如今N多篇文档中然而并没用处。不少童鞋会说咱们能够经过停用词去除掉这些词呀，对。而我说的就是这一类在停用词以外，出现范围很广可是并没有识别用处的词。

(2) spark ml中代码实例

import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}
val wordsData = spark.createDataFrame(Seq(
      "传奇 游戏 战士".split(" "),
      "苹果 梨 香蕉".split(" "),
      "苹果 手机 流畅".split(" ")
    ).map(Tuple1.apply)).toDF("words")
wordsData.show(false)    
// step1 hashingTF
val hashingTF = new HashingTF().setInputCol("words").setOutputCol("rawFeatures").setNumFeatures(2000)
val featurizedData = hashingTF.transform(wordsData)
// step2 计算IDF
val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
val idfModel = idf.fit(featurizedData)
val rescaledData = idfModel.transform(featurizedData)
rescaledData.select("words","features").show(false)

注意：setNumFeatures是设置特征的长度。在这三条数据中，除了苹果外其余的词都仅出现一次，因此能够识别文档的价值都比较大。而苹果同时出如今两条数据中，因此能识别文档的价值被打压，变得很低。

3.词转向量(Word2Vec)

(1) 定义及用途：词转向量是把单词映射到向量空间中，经过一组向量来表明单词。经过计算向量的距离能够表明词的类似度。

(2) spark ml中代码实例

import org.apache.spark.ml.feature.Word2Vec
val documentDF = spark.createDataFrame(Seq(
      "传奇 游戏 战士".split(" "),
      "苹果 梨 香蕉".split(" "),
      "传奇 游戏 种类多".split(" "),
      "苹果 手机 流畅".split(" ")
    ).map(Tuple1.apply)).toDF("text")
val word2Vec = new Word2Vec().setInputCol("text").setOutputCol("result").setVectorSize(10).setMinCount(2)
val model = word2Vec.fit(documentDF)
val result = model.transform(documentDF)
result.show(false)

注意：setVectorSize设置向量的长度。setMinCount设置词在样本中出现的最少次数。好比在上例中咱们设置向量长度为10，至少在两条样本中出现才会转化向量。则知足条件的有"苹果"、“传奇”"游戏"这三个词，因此第一条数据和第三条数据向量距离彻底同样，由于“战士”和“种类多”均只出现一次，并不会用于转化成向量，若是setMinCount设置为1，那么第一条和第三条的向量空间距离会很近，但不会彻底同样，由于也会考虑“战士”和“种类多”这两个词。

3、特征转换

1. 连续型数据转换成离散数据

1.1 二值化(Binarizer)

1. 定义及用途：根据阈值量将连续型数据转化成0-1特征的过程。
2. 注意点：特征值大于阈值将映射为1.0，特征值小于等于阈值将映射为0.0;二值化输入inputCol 支持向量（Vector）和双精度（Double）类型。

1.2 离散化重组(Bucketizer)

1. 定义及用途：根据分段规则将连续型数据转化成各自对应的分段区间内。
2. spark ml中代码实例:

import org.apache.spark.ml.feature.Bucketizer
val data = Array(-8.0, -0.5, -0.3, 0.0, 0.2, 9.0)
val splits = Array(Double.NegativeInfinity, -0.5, 0.0, 0.5, Double.PositiveInfinity)
val dataFrame = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
val bucketizer = new Bucketizer().setInputCol("features").setOutputCol("bucketedFeatures").setSplits(splits)
bucketizer.transform(dataFrame).show(false)

注意点：上例中第三行代码，分段规则制定为(负无穷,0.5),[-0.5,0),[0,0.5),[0.5,正负穷)四段。每一个分段是左闭右开[a,b)方式。当不肯定分裂的上下边界时，应当添加Double.NegativeInfinity和Double.PositiveInfinity以避免越界。

1.3 分位数离散化(QuantileDiscretizer)

1. 定义及用途：根据分位数规则将连续型数据转化成各自对应的分段内。
2. spark ml中代码实例:

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

注意点：setNumBuckets设置分位数分桶数量为3。则将hour数据分红3段。

2.字符串和索引相互转换

2.1 字符串-索引变换(StringIndexer)

1. 定义及用途：将字符串特征转化成索引。不少模型训练过程当中只接受数值特征，因此须要将字符串转换成数值从而进行训练。
2. spark ml中代码实例:

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

注意：索引按照标签频率排序。最多见的标签索引0即表明频率最高的标签。在新数据集中可能会遇到新出现的字符串。好比训练集中只有a,b,c，在新的数据集中会有a,b,c,d。针对新出现的字符串d，有两种策略来处理。 第一种是抛出一个异常（默认状况下），第二种是经过掉用setHandleInvalid(“skip”)来完全忽略包含这类标签的行。

2.2 索引-字符串(IndexToString)

定义及用途：通常是和上面的字符串-索引变换器配套使用。先经过字符串-索引变换器将字符串特征转换成数值类型特征，模型训练完成后经过索引-字符串转换器将数值特征还原成字符串特征。

3. 正则化(Normalizer)

1. 定义及用途：正则化的做用范围是每一行数据，即每一条样本数据。将每一条数据经过计算p-范数进行规范化。正则化操做可使输入数据标准化并提升后期学习算法的效果。
2. spark ml中代码实例：

import org.apache.spark.ml.feature.Normalizer
import org.apache.spark.ml.linalg.{Vector,Vectors}
val data=Seq(Vectors.dense(-1,1,1,8,56),Vectors.dense(-1,3,-1,-9,88),Vectors.dense(0,5,1,10,96), Vectors.dense(0,5,1,11,589),Vectors.dense(0,5,1,11,688))
val df=spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")    
val normalizer = new Normalizer().setInputCol("features").setOutputCol("normFeatures").setP(1.0)
normalizer.transform(df).show(false)

4. 规范化(StandardScaler)

1. 定义及用途：规范化的做用范围是每一列数据，即每一维特征。标准化每一个特征使得其有统一的标准差。一方面，同一个特征中不一样的样本的取值可能会相差很是大，一些异常小或异常大的数据会误导模型的正确训练；另外一方面，若是数据的分布很分散也会影响训练结果。以上两种方式都体如今方差会很是大。
2. spark ml中代码实例：

import org.apache.spark.ml.feature.StandardScaler
import org.apache.spark.ml.linalg.{Vector,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")
val scaler = new StandardScaler().setInputCol("features")
.setOutputCol("scaledFeatures").setWithStd(true).setWithMean(false)
val scalerModel = scaler.fit(dataFrame)
val scaledData = scalerModel.transform(dataFrame)
scaledData.show(false)

注意：上述将每一列的标准差缩放到1。若是特征的标准差为零，则该特征在向量中返回的默认值为0.0。

5. 主成分分析 (PCA)

(1) 定义及用途:主成分分析(PCA)是一种统计学方法，本质是在线性空间中进行一个基变换，使得变换后的数据投影到低维空间的方差最大化。根据变换后方差大小肯定坐标轴的权重或者重要性，权重高的成为主成分。 主要应用于降维。

(2) spark ml中代码实例:

import org.apache.spark.ml.feature.PCA
import org.apache.spark.ml.linalg.{Vector,Vectors}
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")
val scaledDataFrame = new StandardScaler().setInputCol("features").setOutputCol("scaledFeatures").fit(df).transform(df)
val pca = new PCA().setInputCol("features").setOutputCol("pcaFeatures").setK(3).fit(scaledDataFrame)
val pcaDF = pca.transform(scaledDataFrame)
pcaDF.select("features","pcaFeatures").show(false)

注意：经过setK来设置下降到K维空间。上例中原来有5维特征，经过pca下降到3维特征中。pca前必定要对特征向量进行规范化。由于各主成分之间值变化太大，有数量级的差异。标准化特征向量后各主成分之间基本在同一个水平，结果更合理。K值选择问题，能够先选择一个较大的值，经过pcaModel.explainedVariance计算模型的方差，当方差趋于稳定值，选择对应的K值是一个不错的选择。

6. 向量-索引变换(VectorIndexer)

(1) 定义及用途：主要用于批量将离散型特征转化为类别特征

(2) spark ml中代码实例:

import org.apache.spark.ml.feature.VectorIndexer
import org.apache.spark.ml.linalg.Vectors
val data=Seq(Vectors.dense(-1,1,1,8,56),
             Vectors.dense(-1,3,-1,-9,88),
             Vectors.dense(0,5,1,10,96), 
             Vectors.dense(0,5,1,11,589))
val df=spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")    
val indexer = new VectorIndexer().setInputCol("features").setOutputCol("indexed").setMaxCategories(3)
val indexerModel = indexer.fit(df)
indexerModel.transform(df).show(false)

注意：设置setMaxCategories为K，将特征数量小于等于K的特征转化为索引。好比上例中设置setMaxCategories为3，第二列特征有三类，则从新编码为0,1,2。

7. SQL转换器(SQLTransformer)

(1) 定义及用途：不少习惯了使用sql来进行数据处理的童鞋可使用sql转换器处理特征。

(2) spark ml中代码实例:

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

8. 独热编码(OneHotEncoder)

独热编码将标签指标映射为二值变量。

9. 最大值-最小值缩放(MinMaxScaler)

将独立的特征值转换到指定的范围内，一般为[0,1]。

10. 特征向量合并(VectorAssembler)

将原始特征和不一样特征转换器生成的特征合并为单个特征向量。输入列的值将按指定顺序依次添加到一个新向量中。

4、特征选择

特征选择是从特征向量中选择那些更简单有效的特征。适用于在高维数据分析中剔除冗余特征，提高模型的性能。特征选择后的特征是原来特征的一个子集。

1. 向量机(VectorSlicer)

基于已有的特征库，经过索引或者列名来选择部分须要的特征。

2. R公式(RFormula)

经过R模型公式产生一个特征向量和一个标签列。适合在须要作OneHotEncoder的时候，能够一个简单的代码把全部的离散特征转化成数值化表示。

3. 卡方特征选择(ChiSqSelector)

(1) 定义及用途：卡方特征选择根据分类的卡方独立性检验来对特征排序。主要适用于有一堆特征，可是咱们并不知道哪些有用，哪些没用。能够经过卡方特征选择来快速筛选特征。缺点是速度比较慢。

(2) spark ml中代码实例：

import org.apache.spark.ml.feature.ChiSqSelector
import org.apache.spark.ml.feature.VectorIndexer
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.2), 0.0))
val df = spark.createDataset(data).toDF("id", "features", "clicked")
val selector = new ChiSqSelector().setNumTopFeatures(2).setFeaturesCol("features").setLabelCol("clicked").setOutputCol("selectedFeatures")
val result = selector.fit(df).transform(df)
result.show(false)

参考资料：

1. http://spark.apache.org/docs/...
2. http://www.apache.wiki/pages/...
3. https://blog.csdn.net/liuling...

最后，重点来啦~

OPPO商业中心数据标签团队招聘多个岗位，咱们致力于穿透大数据来理解每一个OPPO用户的商业兴趣。数据快速拓展和深挖中，诚邀对数据分析、大数据处理、机器学习/深度学习、NLP等有两年以上经验的您加入咱们，与团队和业务一同成长！

简历投递：ping.wang#oppo.com