摘要：通常在大廠實際項目中會使用Spark來處理大規模數據下的數據挖掘和分析相關工作。本篇從項目實戰中總結常用的Spark特征處理實例，方便小伙伴們更好的使用Spark做數據挖掘相關的工作。

目錄

01 特征處理的意義

02 特征提取

03 特征轉換

04 特征選擇

01 特征處理的意義

在數據挖掘項目中，由于我們獲取的原始數據中包含很多噪聲，所以在真正提供給模型前需要特征處理處理工作，否則再好的模型也只能“Garbage in，garbage out”。

總的來說，特征處理主要包括三部分，特征提取、特征轉換和特征選擇。

02 特征提取

?特征提取一般指從原始數據中抽取特征的過程。

1. 計數向量器(Countvectorizer)

• 定義及用途:計數向量器將所有的文本詞語進行編號，并統計該詞語在文檔中的詞頻作為特征向量。
• 代碼實例

importorg.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 = newCountVectorizer().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)

• 定義及用途：通俗的理解就是計算一個詞區別一篇文檔的程度。通過在一篇文檔中的詞頻和該詞在文檔庫中多少篇文檔中出現綜合來評估。僅通過詞頻來區分一篇文檔是不合理的。比如文檔中會多次出現能代表通用含義的詞，但是這些詞對于文檔的識別并無意義。我們需要的是一些特別的詞，它出現的次數多，并且能在少數的文檔中出現，這些詞才能夠識別文檔。舉個極端的例子，比如“我們”這個詞可能出現在N多篇文檔中然而并沒用處。很多童鞋會說我們可以通過停用詞去除掉這些詞呀，對。而我說的就是這一類在停用詞之外，出現范圍很廣但是并無識別用處的詞。
• 代碼實例

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 = newHashingTF().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)

• 定義及用途：詞轉向量是把單詞映射到向量空間中，通過一組向量來代表單詞。通過計算向量的距離可以代表詞的相似度。
• 代碼實例

?importorg.apache.spark.ml.feature.Word2Vec val documentDF =spark.createDataFrame(Seq("傳奇 游戲 戰士".split(" "), "蘋果 梨 香蕉".split(" "), "傳奇 游戲 種類多".split(" "), "蘋果 手機 流暢".split(" ")).map(Tuple1.apply)).toDF("text") val word2Vec = newWord2Vec().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，那么第一條和第三條的向量空間距離會很近，但不會完全一樣，因為也會考慮“戰士”和“種類多”這兩個詞。

03 特征轉換

1. 連續型數據轉換成離散數據

1.1 二值化(Binarizer)

• 定義及用途：根據閾值量將連續型數據轉化成0-1特征的過程。
• 注意點：特征值大于閾值將映射為1.0，特征值小于等于閾值將映射為0.0;二值化輸入inputCol 支持向量(Vector)和雙精度(Double)類型。

1.2 離散化重組(Bucketizer)

• 定義及用途：根據分段規則將連續型數據轉化成各自對應的分段區間內。
• 代碼實例:

?importorg.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 = newBucketizer().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)

• 定義及用途：根據分位數規則將連續型數據轉化成各自對應的分段內。
• 代碼實例:

?importorg.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 = newQuantileDiscretizer().setInputCol("hour").setOutputCol("result").setNumBuckets(3) val result =discretizer.fit(df).transform(df) result.show()

• 注意點：setNumBuckets設置分位數分桶數量為3。則將hour數據分成3段。

2. 字符串和索引相互轉換

2.1 字符串-索引變換(StringIndexer)

• 定義及用途：將字符串特征轉化成索引。很多模型訓練過程中只接受數值特征，所以需要將字符串轉換成數值從而進行訓練。
• 代碼實例:

importorg.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 = newStringIndexer().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)

• 定義及用途：正則化的作用范圍是每一行數據，即每一條樣本數據。將每一條數據通過計算p-范數進行規范化。正則化操作可以使輸入數據標準化并提高后期學習算法的效果。
• 代碼實例:

?importorg.apache.spark.ml.feature.Normalizer importorg.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)) valdf=spark.createDataFrame(data.map(Tuple1.apply)).toDF("features") val normalizer = newNormalizer().setInputCol("features").setOutputCol("normFeatures").setP(1.0) normalizer.transform(df).show(false)

4. 規范化(StandardScaler)

• 定義及用途：規范化的作用范圍是每一列數據，即每一維特征。標準化每個特征使得其有統一的標準差。一方面，同一個特征中不同的樣本的取值可能會相差非常大，一些異常小或異常大的數據會誤導模型的正確訓練；另一方面，如果數據的分布很分散也會影響訓練結果。以上兩種方式都體現在方差會非常大。
• 代碼實例:

?importorg.apache.spark.ml.feature.StandardScaler importorg.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 = newStandardScaler().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)

• 定義及用途:主成分分析(PCA)是一種統計學方法，本質是在線性空間中進行一個基變換，使得變換后的數據投影到低維空間的方差最大化。根據變換后方差大小確定坐標軸的權重或者重要性，權重高的成為主成分。主要應用于降維。
• 代碼實例:

importorg.apache.spark.ml.feature.PCA importorg.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 = newStandardScaler().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)

• 定義及用途：主要用于批量將離散型特征轉化為類別特征
• 代碼實例:

importorg.apache.spark.ml.feature.VectorIndexer importorg.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)) valdf=spark.createDataFrame(data.map(Tuple1.apply)).toDF("features") val indexer = newVectorIndexer().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)

• 定義及用途：很多習慣了使用sql來進行數據處理的童鞋可以使用sql轉換器處理特征。
• 代碼實例:

?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 = newSQLTransformer().setStatement( "SELECT *, (v1 + v2) AS v3, (v1 * v2) AS v4 FROM __THIS__") sqlTrans.transform(df).show()

8. 獨熱編碼(OneHotEncoder):獨熱編碼將標簽指標映射為二值變量。

9. 最大值-最小值縮放(MinMaxScaler):將獨立的特征值轉換到指定的范圍內，通常為[0,1]。

10. 特征向量合并(VectorAssembler)：將原始特征和不同特征轉換器生成的特征合并為單個特征向量。輸入列的值將按指定順序依次添加到一個新向量中。

04 特征選擇

特征選擇是從特征向量中選擇那些更簡單有效的特征。適用于在高維數據分析中剔除冗余特征，提升模型的性能。特征選擇后的特征是原來特征的一個子集。

1. 向量機(VectorSlicer) ：基于已有的特征庫，通過索引或者列名來選擇部分需要的特征。

2. R公式(RFormula) ：通過R模型公式產生一個特征向量和一個標簽列。適合在需要做OneHotEncoder的時候，可以一個簡單的代碼把所有的離散特征轉化成數值化表示。

3. 卡方特征選擇(ChiSqSelector)

• 定義及用途：卡方特征選擇根據分類的卡方獨立性檢驗來對特征排序。主要適用于有一堆特征，但是我們并不知道哪些有用，哪些沒用。可以通過卡方特征選擇來快速篩選特征。缺點是速度比較慢。
• 代碼實例:

importorg.apache.spark.ml.feature.ChiSqSelector importorg.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 = newChiSqSelector().setNumTopFeatures(2).setFeaturesCol("features").setLabelCol("clicked").setOutputCol("selectedFeatures") val result =selector.fit(df).transform(df) result.show(false)

?

參考資料

[1] http://spark.apache.org/docs/latest/ml-features.html

[2] http://www.apache.wiki/pages/viewpage.action?pageId=5505205

[3] https://blog.csdn.net/liulingyuan6/article/details/53413728

特別說明：本篇主要參考資料是Spark官網資料，結合實際項目實例總結出常用的特征處理的實例。有興趣的小伙伴可以參考上面的資料[1]。

?

喜歡本類型文章的小伙伴可以關注我的微信公眾號：數據拾光者。有任何干貨我會首先發布在微信公眾號，還會同步在知乎、頭條、簡書、csdn等平臺。也歡迎小伙伴多交流。如果有問題，可以在微信公眾號隨時Q我哈。

總結

以上是生活随笔為你收集整理的spark 不同模式用途_Spark中那些常用的特征处理操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。