資料探勘演算法中有很大一部分都是資料預處理工作，畢竟現有模型都是比較成熟的，只需要學會呼叫就好，如何把原始資料轉化為演算法模型適用的資料結構也是很重要的一步。spark ML中提供了對特徵的提取（Extracting），轉換（transforming）和選擇（selecting）工具。

• 特徵提取：從原始資料中提取特徵
• 特徵轉換：特徵的擴充套件，特徵的轉化，特徵的修改
• 特徵選擇：從大規模特徵集中選取一個子集

特徵提取

TF-IDF

“詞頻－逆向檔案頻率”（TF-IDF）是一種在文字挖掘中廣泛使用的特徵向量化方法，它可以體現一個文件中詞語在語料庫中的重要程度。

詞頻TF(t,d)是詞語t在文件d中出現的次數。檔案頻率DF(t,D)是包含詞語的文件的個數。如果我們只使用詞頻來衡量重要性，很容易過度強調在文件中經常出現，卻沒有太多實際資訊的詞語，比如“a”，“the”以及“of”。如果一個詞語經常出現在語料庫中，意味著它並不能很好的對文件進行區分。TF-IDF就是在數值化文件資訊，衡量詞語能提供多少資訊以區分文件。定義如下：

​ 此處 |D| 是語料庫中總的文件數。公式中使用log函式，當詞出現在所有文件中時，它的IDF值變為0。加1是為了避免分母為0的情況。TF-IDF 度量值表示如下：

在Spark ML庫中，TF-IDF被分成兩部分：TF(HashingTF) 和 IDF

• HashingTF 是一個Transformer，在文字處理中，接收詞條的集合然後把這些集合轉化成固定長度的特徵向量。這個演算法在雜湊的同時會統計各個詞條的詞頻。
• 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重新調整特徵向量。這種轉換通常可以提高使用文字特徵的效能。

在下面的程式碼中，我們以一組句子為例，使用Tokenizer將每一條句子分解為單詞，對每一條句子（詞袋），我們使用HashingTF 將其轉化為特徵向量，最後使用IDF 重新調整特徵向量。

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

// 建立一個簡單的DataFrame，每一個句子代表一個文件
val sentenceData = spark.createDataFrame(Seq(
  (0.0, "Hi I heard about Spark"),
  (0.0, "I wish Java could use case classes"),
  (1.0, "Logistic regression models are neat")
)).toDF("label", "sentence")

// 在得到文件集合後，即可用tokenizer對句子進行分詞。
val tokenizer = new Tokenizer().setInputCol("sentence").setOutputCol("words")
val wordsData = tokenizer.transform(sentenceData)
wordsData.show(false)  // 如果不設定為false的話，將會省略掉20個以後的字元

/*
+-----+-----------------------------------+------------------------------------------+
|label|sentence                           |words                                     |
+-----+-----------------------------------+------------------------------------------+
|0.0  |Hi I heard about Spark             |[hi, i, heard, about, spark]              |
|0.0  |I wish Java could use case classes |[i, wish, java, could, use, case, classes]|
|1.0  |Logistic regression models are neat|[logistic, regression, models, are, neat] |
+-----+-----------------------------------+------------------------------------------+

*/

// 使用HashingTF的transform()方法把句子雜湊成特徵向量，這裡設定雜湊表的桶數為2000。
val hashingTF = new HashingTF()
  .setInputCol("words").setOutputCol("rawFeatures").setNumFeatures(20)

val featurizedData = hashingTF.transform(wordsData)

featurizedData.show(false)
/*
可以看到，分詞序列被變換成一個稀疏特徵向量，其中每個單詞都被雜湊成了一個不同的索引值，特徵向量在某一維度上的值即該詞彙在文件中出現的次數

+-----------------------------------------+
|rawFeatures                              |
+-----------------------------------------+
|(20,[5,6,9],[2.0,1.0,2.0])               |
|(20,[3,5,12,14,18],[2.0,2.0,1.0,1.0,1.0])|
|(20,[5,12,14,18],[1.0,2.0,1.0,1.0])      |
+-----------------------------------------+

*/

// 最後，使用IDF來對單純的詞頻特徵向量進行修正，使其更能體現不同詞彙對文字的區別能力，IDF是一個Estimator，呼叫fit()方法並將詞頻向量傳入，即產生一個IDFModel。
val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
val idfModel = idf.fit(featurizedData)

// 很顯然，IDFModel是一個Transformer，呼叫它的transform()方法，即可得到每一個單詞對應的TF-IDF度量值。
val rescaledData = idfModel.transform(featurizedData)
rescaledData.select("label", "features").take(3).foreach(println)

/*
可以看到，特徵向量已經被其在語料庫中出現的總次數進行了修正，通過TF-IDF得到的特徵向量，在接下來可以被應用到相關的機器學習方法中

[0.0,(20,[5,6,9],[0.0,0.6931471805599453,1.3862943611198906])]
[0.0,(20,[3,5,12,14,18],[1.3862943611198906,0.0,0.28768207245178085,0.28768207245178085,0.28768207245178085])]
[1.0,(20,[5,12,14,18],[0.0,0.5753641449035617,0.28768207245178085,0.28768207245178085])]
*/

特徵轉換

值得注意的是，用於特徵轉換的轉換器和其他的機器學習演算法一樣，也屬於ML Pipeline模型的一部分，可以用來構成機器學習流水線，以StringIndexer為例，其儲存著進行標籤數值化過程的相關 超引數，是一個Estimator，對其呼叫fit(..)方法即可生成相應的模型StringIndexerModel類，很顯然，它儲存了用於DataFrame進行相關處理的 引數，是一個Transformer（其他轉換器也是同一原理）

Tokenizer

根據原始碼中的解釋 ：

A tokenizer that converts the input string to lowercase and then splits it by white spaces.

tokenizer 主要有兩個功能，1. 將詞語轉為小寫字母 2. 按空格劃分字串

注： 如果想使用其他規則劃分字串，可以使用 RegexTokenizer

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 = new Tokenizer().setInputCol("sentence").setOutputCol("words")
val regexTokenizer = new RegexTokenizer()
  .setInputCol("sentence")
  .setOutputCol("words")
  .setPattern("\\W") // alternatively .setPattern("\\w+").setGaps(false)

// 自定義一個函式
val countTokens = udf { (words: Seq[String]) => words.length }

val tokenized = tokenizer.transform(sentenceDataFrame)
val regexTokenized = regexTokenizer.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     |
+-----------------------------------+------------------------------------------+------+
*/

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     |
+-----------------------------------+------------------------------------------+------+
*/

很明顯的可以看到二者的區別，tokenized是將非空格分割的字串當做 一個整體。

Binarizer

連續特徵根據閾值二值化，大於閾值的為1.0，小於等於閾值的為0.0。二值化是機器學習中很常見的思路，可以將連續型資料轉化為離散型。

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

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|
+---+-------+-----------------+
*/

StringIndexer

StringIndexer轉換器可以把一列類別型的特徵（或標籤）進行編碼，使其數值化，索引的範圍從0開始，該過程可以使得相應的特徵索引化，使得某些無法接受類別型特徵的演算法可以使用，並提高諸如決策樹等機器學習演算法的效率。

索引構建的順序為標籤的頻率，優先編碼頻率較大的標籤，所以出現頻率最高的標籤為0號。
如果輸入的是數值型的，我們會把它轉化成字元型，然後再對其進行編碼。直接上程式碼。

首先建立一個簡單的DataFrame，它只包含一個id列和一個標籤列category：

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

df.show(false)

/*
+---+--------+
|id |category|
+---+--------+
|0  |a       |
|1  |b       |  
|2  |c       |
|3  |a       |
|4  |a       |
|5  |c       |
+---+--------+
*/

隨後，我們建立一個StringIndexer物件，設定輸入輸出列名，其餘引數採用預設值，並對這個DataFrame進行訓練，產生StringIndexerModel物件, 最後利用該物件對DataFrame進行轉換操作。可以看到，StringIndexerModel依次按照出現頻率的高低，把字元標籤進行了排序，即出現最多的“a”被編號成0，“c”為1，出現最少的“b”為2。

val indexer = new StringIndexer()
  .setInputCol("category")
  .setOutputCol("categoryIndex")

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

考慮這樣一種情況，我們使用已有的資料構建了一個StringIndexerModel，然後再構建一個新的DataFrame，這個DataFrame中有著模型內未曾出現的標籤“d”，用已有的模型去轉換這一DataFrame會有什麼效果？

實際上，如果直接轉換的話，Spark會丟擲異常，報出“Unseen label: d”的錯誤。

為了處理這種情況，在模型訓練後，可以通過設定setHandleInvalid(“skip”)來忽略掉那些未出現的標籤，這樣，帶有未出現標籤的行將直接被過濾掉，

測試一下，我們建立一個df2，包含df未出現的字元d，用df訓練的模型來轉換df2。

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

indexed.transform(df2).show(false)

報錯：org.apache.spark.SparkException: Unseen label: d.

indexed.setHandleInvalid("skip").transform(df2).show(false)

/*
+---+--------+-------------+
|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          |
+---+--------+-------------+

*/

IndexToString

與StringIndexer相對應，IndexToString的作用是把標籤索引的一列重新映射回原有的字元型標籤，一般都是和StringIndexer配合使用，先用StringIndexer將原有標籤轉化成標籤索引，進行模型訓練，然後在預測標籤的時候再把標籤索引轉化成原有的字元標籤。
我們還是先來建立個DataFrame

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

然後使用StringIndexer標籤索引化。

val indexer = new StringIndexer()
  .setInputCol("category")
  .setOutputCol("categoryIndex")
  .fit(df)
val indexed = indexer.transform(df)
/*
+---+--------+-------------+
|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          |
+---+--------+-------------+
*/

緊接著，使用IndexToString還原標籤索引

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

val converted = converter.transform(indexed)

/*
+---+--------+-------------+----------------+
|id |category|categoryIndex|originalCategory|
+---+--------+-------------+----------------+
|0  |a       |0.0          |a               |
|1  |b       |2.0          |b               |
|2  |c       |1.0          |c               |
|3  |a       |0.0          |a               |
|4  |a       |0.0          |a               |
|5  |c       |1.0          |c               |
+---+--------+-------------+----------------+
*/

OneHotEncoder

獨熱編碼是指把一列標籤索引對映成一列二進位制陣列，且最多的時候只有一位有效。這種編碼適合一些期望類別特徵為連續特徵的演算法（迴歸），個人感覺獨熱編碼比較適用，其原理這裡就不說了，給大家一個連線：
http://blog.csdn.net/bitcarmanlee/article/details/51472816
與sklearn 不同之處在於sparkml中使用SparseVector來表示。

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

val df = sqlContext.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")
val encoded = encoder.transform(indexed)
encoded.select("id", "categoryVec").show()

/*
+---+--------+-------------+-------------+
| id|category|categoryIndex|  categoryVec|
+---+--------+-------------+-------------+
|  0|       a|          0.0|(2,[0],[1.0])|
|  1|       b|          2.0|    (2,[],[])|
|  2|       c|          1.0|(2,[1],[1.0])|
|  3|       a|          0.0|(2,[0],[1.0])|
|  4|       a|          0.0|(2,[0],[1.0])|
|  5|       c|          1.0|(2,[1],[1.0])|
+---+--------+-------------+-------------+
*/

將categoryVec轉換成編碼模式：a -> 10 , c ->01 ,至於b為空，是應為spark預設不包含最後一個類別，OneHotEncoder.setDropLast(false)可以設定是否包含最後一個類別。

VectorIndexer

VectorIndexer解決向量資料集中的類別特徵索引。它可以自動識別哪些特徵是類別型(離散)的，並且將原始值轉換為類別索引。它的處理流程如下：

​ 1. 獲得一個向量型別的輸入以及maxCategories引數。

​ 2. 基於不同特徵值的數量來識別哪些特徵需要被類別化，不同特徵值的數量小於maxCategories，則這個特徵需要被類別化（離散化），否則視為連續值，不做改變。

​ 3.對於每一個類別特徵計算0-based（從0開始）類別索引。

索引後的類別特徵可以幫助決策樹等演算法恰當的處理類別型特徵，並得到較好結果。

我們讀入一個數據集，然後使用VectorIndexer來決定哪些特徵需要被作為類別特徵，將類別特徵轉換為他們的索引。

import org.apache.spark.ml.feature.VectorIndexer
val data = Seq(Vectors.dense(-1.0, 1.0, 1.0),Vectors.dense(-1.0, 3.0, 1.0), Vectors.dense(0.0, 5.0, 1.0))
val df = sqlContext.createDataFrame(data.map(Tuple1.apply)).toDF("features")
val indexer = new VectorIndexer()
         .setInputCol("features")
         .setOutputCol("indexed")
         .setMaxCategories(2)
val indexerModel = indexer.fit(df)
val indexedData = indexerModel.transform(df)
 indexedData.show()
 /*
+--------------+-------------+
|      features|      indexed|
+--------------+-------------+
|[-1.0,1.0,1.0]|[1.0,1.0,0.0]|
|[-1.0,3.0,1.0]|[1.0,3.0,0.0]|
| [0.0,5.0,1.0]|[0.0,5.0,0.0]|
+--------------+-------------+
*/

從上例可以看到，我們設定maxCategories為2，即只有種類小於2的特徵才被認為是類別型特徵，否則被認為是連續型特徵。其中類別型特徵將被進行編號索引，為了索引的穩定性，規定如果這個特徵值為0，則一定會被編號成0，這樣可以保證向量的稀疏度。於是，我們可以看到第0類和第2類的特徵由於種類數不超過2，被劃分成類別型特徵，並進行了索引，且為0的特徵值也被編號成了0號。

Normalizer

Normalizer是將資料集的每一行資料歸一化的轉換器，一行資料看做一個向量，它帶一個引數P（預設值是2），歸一化就是向量中的每個元素除以向量的範數（P規定使用哪一個範數）。

val data = Seq(Vectors.dense(-1.0, 1.0, 1.0),Vectors.dense(-1.0, 3.0, 1.0), Vectors.dense(0.0, 5.0, 1.0))
val df = sqlContext.createDataFrame(data.map(Tuple1.apply)).toDF("features")
val normalizer = new Normalizer()
  .setInputCol("features")
  .setOutputCol("normFeatures")
  .setP(2.0)

val l1NormData = normalizer.transform(dataFrame)
l1NormData.show()
/*
+--------------+-------------------------------------------------------------+
|features      |normFeatures                                                 |
+--------------+-------------------------------------------------------------+
|[-1.0,1.0,1.0]|[-0.5773502691896258,0.5773502691896258,0.5773502691896258]  |
|[-1.0,3.0,1.0]|[-0.30151134457776363,0.9045340337332909,0.30151134457776363]|
|[0.0,5.0,1.0] |[0.0,0.9805806756909202,0.19611613513818404]                 |
+--------------+-------------------------------------------------------------+
*/

我們設定的P為2，則使用L2範數來歸一化向量，L2範數是指向量各元素的平方和然後求平方根，範數的相關資料可以自行百度，針對[-1,1,1],他的L2範數為3√, 所以歸一化後的向量為[−13√,13√,13√]

val lInfNormData = normalizer.transform(dataFrame, normalizer.p -> Double.PositiveInfinity)
lInfNormData.show(false)

/*
+--------------+--------------------------------------------+
|features      |normFeatures                                |
+--------------+--------------------------------------------+
|[-1.0,1.0,1.0]|[-1.0,1.0,1.0]                              |
|[-1.0,3.0,1.0]|[-0.3333333333333333,1.0,0.3333333333333333]|
|[0.0,5.0,1.0] |[0.0,1.0,0.2]                               |
+--------------+--------------------------------------------+
*/

無窮範數就是取向量中的最大值，所以使用無窮範數歸一化後的向量就是原向量中元素除以向量中的最大值。

VectorAssembler

VectorAssembler是非常實用的一個轉換器，它的作用很簡單，就是把DataFrame中的若干列合併為一個向量列，可類似於Excel中的合併單元格。有了VectorAssembler之後，除了使用反射機制建立DataFrame之外，我們又有了一種新思路，即通過普通方式建立DataFrame，再使用VectorAssembler創造自己所需要的資料格式。

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

val dataset = sqlcontext.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)
output.show(false)

/*
+---+----+------+--------------+-------+-----------------------+
|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]|
+---+----+------+--------------+-------+-----------------------+
*/

特徵選擇

VectorSlicer

VectorSlicer是一個切片工具，和VectorAssembler相對應，它的作用就是將一個向量切片，選擇向量的子集。它提供了兩種切片方式。

• setIndices(), 按照向量的下標索引切片（從0開始）
• setNames()，按照列名切片

import org.apache.spark.ml.attribute.{Attribute, AttributeGroup, NumericAttribute}
import org.apache.spark.ml.feature.VectorSlicer
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType

val data = Array(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 dataRDD = sc.parallelize(data)
val dataset = sqlContext.createDataFrame(dataRDD, 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)
ouput.show(false)
/*
+--------------+---------+
|userFeatures  |features |
+--------------+---------+
|[-2.0,2.3,0.0]|[2.3,0.0]|
+--------------+---------+
*/