1 TF-IDF

　　TF-IDF是一種特徵向量化方法，這種方法多用於文字挖掘，通過演算法可以反應出詞在語料庫中某個文件中的重要性。文件中詞記為t，文件記為d , 語料庫記為D . 詞頻TF(t,d) 是詞t 在文件d 中出現的次數。文件頻次DF(t,D) 是語料庫中包括詞t的文件數。如果使用詞在文件中出現的頻次表示詞的重要程度，那麼很容易取出反例，即有些詞出現頻率高反而沒多少資訊量， 如,”a” , “the” , “of” 。如果一個詞在語料庫中出現頻率高，說明它在特定文件集中資訊量很低。逆文件頻次（inverse document frequency）是詞所能提供的資訊量的一種度量：

此處|D| 是語料庫中總的文件數，注意到，公式中使用log函式，當詞出現在所有文件中時，它的IDF值變為0. 給IDF加一個防止在此情況下分母為0. TF-IDF 度量值表示如下：

對於TF 和 IDF 定義有多種，spark.mllib 中，分開定義TF 和IDF 。

　　Spark.mllib 中實現詞頻率統計使用特徵hash的方式，原始的特徵通過hash函式，對映到一個索引值。後面只需要統計這些索引值的頻率，就可以知道對應詞的頻率。這種方式避免設計一個全域性1對1的詞到索引的對映，這個對映在對映大量語料庫時需要花費更長的時間。但需要注意，通過hash的方式可能會對映到同一個值的情況，即不同的原始特徵通過Hash對映後是同一個值。為了降低這種情況出現的概率，我們只能對特徵向量升維。i.e., hash表的桶數，預設特徵維度是 2^20 = 1,048,576.

注意：spark.mllib 不支援文字分段，詳見 Stanford nlp group http://nlp.stanford.edu/和 scalanlp/chalk : https://github.com/scalanlp/chalk
　　TF實際是統計詞hash之後索引值的頻次，可使用HashingTF 方法並傳入RDD[Iterable[_]] ， IDF 需要使用IDF方法。需要注意，每條記錄是可iterable的字串或其它型別。

import org.apache.spark.rdd.RDD
import org.apache.spark.<span class="wp_keywordlink_affiliate"><a href="http://www.iteblog.com/archives/tag/spark" title="" target="_blank" data-original-title="View all posts in Spark">Spark</a></span>Context
import org.apache.spark.mllib.feature.HashingTF
import org.apache.spark.mllib.linalg.Vector

val sc: <span class="wp_keywordlink_affiliate"><a href="http://www.iteblog.com/archives/tag/spark" title="" target="_blank" data-original-title="View all posts in Spark">Spark</a></span>Context = ...

// Load documents (one per line).
val documents: RDD[Seq[String]] = sc.textFile("...").map(_.split(" ").toSeq)

val hashingTF = new HashingTF()
val tf: RDD[Vector] = hashingTF.transform(documents)
 

HashingTF 方法只需要一次資料互動，而IDF需要兩次資料互動：第一次計算IDF向量，第二次需要和詞頻次相乘

import org.apache.spark.mllib.feature.IDF

// ... continue from the previous example
tf.cache()
val idf = new IDF().fit(tf)
val tfidf: RDD[Vector] = idf.transform(tf)

spark.mllib支援乎略詞頻低於文件最小數，需要把minDocFreq這個數傳給IDF構架函式。在此情況下，對應的IDF值設定為0，

import org.apache.spark.mllib.feature.IDF

// ... continue from the previous example
tf.cache()
val idf = new IDF(minDocFreq = 2).fit(tf)
val tfidf: RDD[Vector] = idf.transform(tf)

2 Word2Vect (詞到向量)

　　Word2Vec 計算詞表徵向量的分佈，這樣可以利用相似相近的詞表徵分佈在鄰近的向量空間，好處就是易於產生新型模型，且模型預測的誤差也容易解釋。向量分佈在自然語言處理中是很有用的，特定像命名實體識別，歧義消除，句法分析，詞性標記和機器翻譯。

2.1 模型

　　Word2vec 的實現中，我們使用skip-gram模型。Skip-gram的訓練目標是學習詞表徵向量分佈，這個分佈可以用來預測句子所在的語鏡。數學上，給定一組訓練詞w_1, … w_T ,skip-gram模型的目標是最大化平均log-似然。

此處 k 是訓練樣本視窗。

　　在skip-gram模型中，每個單詞w 關聯兩個向量u_w 和v_w ，其中u_w是單詞w的向量表示，v_w是單詞對應的語境。對於給定的單詞w_j ，計算預測結果的正確概率由以下softmax 模型。

2.2 例子

　　下例子列舉如何載入文字檔案，將文字內容存放到RDD[Seq[String]]，從RDD構造一個Word2Vec例項，將輸入資料送入此例項訓練得到Word2VecModel模型。最終，我們展示特定詞的前40個同義詞。為了執行這個例子，首先下載text8(http://mattmahoney.net/dc/text8.zip) 資料，解壓到特定的目錄下。此處我們假設解壓出來的檔案還叫text8 ，並且在當前目錄。

import org.apache.spark._
import org.apache.spark.rdd._
import org.apache.spark.SparkContext._
import org.apache.spark.mllib.feature.{Word2Vec, Word2VecModel}

val input = sc.textFile("text8").map(line => line.split(" ").toSeq)

val word2vec = new Word2Vec()

val model = word2vec.fit(input)

val synonyms = model.findSynonyms("china", 40)

for((synonym, cosineSimilarity) <- synonyms) {
  println(s"$synonym $cosineSimilarity")
}

// Save and load model
model.save(sc, "myModelPath")
val sameModel = Word2VecModel.load(sc, "myModelPath")

3 standardscaler標準化

　　標準化是通過變化將原始資料放縮到單位方差，通過平移資料得到均值為0（如果原資料均值不為0，需要對取樣資料求出樣本均值，將原始資料減雲樣本均值，即得到均值為0的新資料）。

　　例如，支援向量機的RBF 核，或L1和L2空間的正則線性模型，這兩個例子很能說明問題，經過標準化所有特徵的計算能得到更好的結果。

　　標準化後的資料，在最優化過程中會更快的收斂，同時也會在模型訓練時防止方差大的資料對整體資料的影響。

3.1 模型擬合

　　標準化需要配置以下引數：

　　1 withMean 預設是假(false)。在標準化之前將原始資料以均值為中心，這樣會使標準化後的資料分佈相對緊密些，這種方法不適合於稀鬆的資料集，否則會觸發異常。
　　2 withStd 預設是真(true) , 意味將資料標準化到單位方差。

　　在StandardScaler 中提供一個擬合方法將RDD[Vector]作為輸入，學習輸入的統計資訊，將輸入集合變換成單位標準差，變換結果可能（也可能不是）均值為0 ，通過配置StandardScaler 來實現。

　　模型支援VectorTransformer ，可以將標準向量變換成新的向量，或者將RDD[Vector] 變換到新的RDD[Vector]。

　　如果特徵向量某個維度的方差為0，則特徵向量這個維度的變換結果仍然是0.0

3.2 例子

　　下例展示如何載入libsvm格式資料，將資料標準化後得到新的向量，此新向量的標準差是1，均值可能（也可能不是） 0 。

import org.apache.spark.SparkContext._
import org.apache.spark.mllib.feature.StandardScaler
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.util.MLUtils

val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")

val scaler1 = new StandardScaler().fit(data.map(x => x.features))
val scaler2 = new StandardScaler(withMean = true, withStd = true).fit(data.map(x => x.features))
// scaler3 is an identical model to scaler2, and will produce identical transformations
val scaler3 = new StandardScalerModel(scaler2.std, scaler2.mean)

// data1 will be unit variance.
val data1 = data.map(x => (x.label, scaler1.transform(x.features)))

// Without converting the features into dense vectors, transformation with zero mean will raise
// exception on sparse vector.
// data2 will be unit variance and zero mean.
val data2 = data.map(x => (x.label, scaler2.transform(Vectors.dense(x.features.toArray))))

4 正規化


4.1 例子

import org.apache.spark.SparkContext._
import org.apache.spark.mllib.feature.Normalizer
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.util.MLUtils

val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")

val normalizer1 = new Normalizer()
val normalizer2 = new Normalizer(p = Double.PositiveInfinity)

// Each sample in data1 will be normalized using $L^2$ norm.
val data1 = data.map(x => (x.label, normalizer1.transform(x.features)))

// Each sample in data2 will be normalized using $L^\infty$ norm.
val data2 = data.map(x => (x.label, normalizer2.transform(x.features)))

5 ChiSqSelector(ChiSq選擇器)

　　在模型構造階段，特徵選擇從特徵向量中剔除相關的維度，即對特徵空間進行降維，這樣可以加速迭代過程，並提升學習效率。

　　ChiSqSelector 實現基於chi-squared 的特徵選擇器，它處理歸類特徵的類標籤，ChiSqSelector 基於Chi-Squared 檢驗對特徵進行排序，而不直接考慮特徵向量的類別，選取排序靠前的特徵向量，因為這些特徵向量能很好的決定類別標籤。這就好比選取對分類有決定意義的特徵向量。

　　在實際中，選取檢驗集可以優化特徵的數量。

5.1 模型擬合

　　ChiSqSelector 演算法配置 numTopFeatures 引數來確定選取排名前多少個特徵向量。

　　擬合方法的輸入是歸類特徵的RDD[LabeledPoint]，通過學習統計資訊，返回ChiSqSelectorModel模型，這個模型可以用於對特徵空間進行降維。這個模型可以處理輸入Vector,得到降維後的Vector , 或者對RDD[Vector]進行降維。

　　當然，也可以構造一個特徵索引（索引按升序排列）, 對這個索引的陣列訓練ChiSqSelectorModel模型。

5.2例子

　　下例展現ChiSqSelector的基礎應用，輸入矩陣的每個元素的範圍 0 ~ 255 。

import org.apache.spark.SparkContext._
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.mllib.feature.ChiSqSelector

// Load some data in libsvm format
val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")
// Discretize data in 16 equal bins since ChiSqSelector requires categorical features
// Even though features are doubles, the ChiSqSelector treats each unique value as a category
val discretizedData = data.map { lp =>
  LabeledPoint(lp.label, Vectors.dense(lp.features.toArray.map { x => (x / 16).floor } ) )
}
// Create ChiSqSelector that will select top 50 of 692 features
val selector = new ChiSqSelector(50)
// Create ChiSqSelector model (selecting features)
val transformer = selector.fit(discretizedData)
// Filter the top 50 features from each feature vector
val filteredData = discretizedData.map { lp => 
  LabeledPoint(lp.label, transformer.transform(lp.features)) 
}

6 Hadamard乘積(ElementwiseProduct)

　　ElementwiseProduct對輸入向量的每個元素乘以一個權重向量的每個元素，對輸入向量每個元素逐個進行放縮。這個稱為對輸入向量v 和變換向量scalingVec 使用Hadamard product(阿達瑪積)進行變換，最終產生一個新的向量。用向量 w 表示 scalingVec ，則Hadamard product可以表示為

Hamard 乘積需要配置一個權向量 scalingVec

　　1) scalingVec 變換向量

　　ElementwiseProduct實現 VectorTransformer 方法，就可以對向量乘以權向量，得到新的向量，或者對RDD[Vector] 乘以權向量得到RDD[Vector]

6.1 例子

　　下例展示如何對向量進行ElementwiseProduct變換

import org.apache.spark.SparkContext._
import org.apache.spark.mllib.feature.ElementwiseProduct
import org.apache.spark.mllib.linalg.Vectors

// Create some vector data; also works for sparse vectors
val data = sc.parallelize(Array(Vectors.dense(1.0, 2.0, 3.0), Vectors.dense(4.0, 5.0, 6.0)))

val transformingVector = Vectors.dense(0.0, 1.0, 2.0)
val transformer = new ElementwiseProduct(transformingVector)

// Batch transform and per-row transform give the same results:
val transformedData = transformer.transform(data)
val transformedData2 = data.map(x => transformer.transform(x))

7 PCA

　　PCA可以將特徵向量投影到低維空間，實現對特徵向量的降維。

7.1 例子

　　下例展示如何計算特徵向量空間的主成分，使用主成分對向量投影到低維空間，同時保留向量的類標籤。

import org.apache.spark.mllib.regression.LinearRegressionWithSGD
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.feature.PCA

val data = sc.textFile("data/mllib/ridge-data/lpsa.data").map { line =>
  val parts = line.split(',')
  LabeledPoint(parts(0).toDouble, Vectors.dense(parts(1).split(' ').map(_.toDouble)))
}.cache()

val splits = data.randomSplit(Array(0.6, 0.4), seed = 11L)
val training = splits(0).cache()
val test = splits(1)

val pca = new PCA(training.first().features.size/2).fit(data.map(_.features))
val training_pca = training.map(p => p.copy(features = pca.transform(p.features)))
val test_pca = test.map(p => p.copy(features = pca.transform(p.features)))

val numIterations = 100
val model = LinearRegressionWithSGD.train(training, numIterations)
val model_pca = LinearRegressionWithSGD.train(training_pca, numIterations)

val valuesAndPreds = test.map { point =>
  val score = model.predict(point.features)
  (score, point.label)
}

val valuesAndPreds_pca = test_pca.map { point =>
  val score = model_pca.predict(point.features)
  (score, point.label)
}

val MSE = valuesAndPreds.map{case(v, p) => math.pow((v - p), 2)}.mean()
val MSE_pca = valuesAndPreds_pca.map{case(v, p) => math.pow((v - p), 2)}.mean()

println("Mean Squared Error = " + MSE)
println("PCA Mean Squared Error = " + MSE_pca)