PySpark Feature Tool

1. 數據準備

我們定義了一些測試數據，方便驗證函數的有效性；同時對于大多數初學者來說，明白函數的輸入是什么，輸出是什么，才能更好的理解特征函數和使用特征：

df = spark.createDataFrame([('zhu', "Hi I heard about pySpark"),('xiang', "I wish python could use case classes"),('yu', "Logistic regression models are neat") ], ["id", "sentence"]) # functionTestData +-----+------------------------------------+ |id |sentence | +-----+------------------------------------+ |zhu |Hi I heard about pySpark. | |xiang|I wish python could use case classes| |yu |Logistic regression models are neat | +-----+------------------------------------+

2.數據讀取

# !/usr/bin/env python # -*- coding: utf-8 -*-######################################################################################################################## # Creater : Zhu Xiangyu.DOTA # Creation Time : 2020-2-17 12:45:09 # Description : PySpark 特征工程工具集 # Modify By : # Modify Time : # Modify Content : # Script Version : 2.0.0.9 ########################################################################################################################import math from pyspark.sql import SparkSession spark = SparkSession.builder.appName('DOTAd_Features_Tool').enableHiveSupport().getOrCreate() spark.conf.set("spark.sql.shuffle.partitions", 1000) spark.conf.set("spark.default.parallelism", 2000)def get_params():return {# Function Can be Used'column1' : "TFIDF", # 詞頻－逆向文件頻率'column2' : "Word2Vec",'column3' : "CountVectorizer",'column4' : "OneHotEncoder",'column5' : "StringIndexer",'column6' : "IndexToString",'column7' : "PCA",'column8' : "Binarizer",'column9' : "Tokenizer",'column10': "StopWordsRemover", #'column11': "NGram", #'column12': "DCT", # 離散余弦變換'column13': "ChiSqSelector", # 卡方校驗'column14': "PearsonCorr", # 皮爾遜系數}def main():# Reset params######################################################################################## 庫名.表名dataset_Name = ""dataset = spark.sql("select * from {dataset_Name}".format(dataset_Name = dataset_Name)).fillna(0)## 結果存儲目標 庫名.表名saveAsTable_Name = ""## 指定對列col進行function操作 {col:function}params = {'sentence': "TFIDF"}######################################################################################### functionTestDatadf = spark.createDataFrame([('zhu', "Hi I heard about pySpark"),('xiang', "I wish python could use case classes"),('yu', "Logistic regression models are neat")], ["id", "sentence"])# Feature Transformfeatures = featureTool(dataset,params) # Test-Model : dataset = dffeatures.show(5)# Save Feature as tablesaveResult(features,saveAsTable_Name)

3.數據存儲

# SaveTableAs def saveResult(res,saveAsTable_Name='dota_tmp.dota_features_tool_save_result', saveFormat="orc",saveMode="overwrite"):res.write.saveAsTable(name=saveAsTable_Name, format=saveFormat,mode=saveMode)

4.特征函數

def featureTool(df,params):dataCols,targetCols = df.columns,params.keys()exeColumns = list(params.keys())[0]exeDefFunction = params[exeColumns]print(exeColumns+"-->"+exeDefFunction+"(df,{exeColumns})".format(exeColumns=exeColumns))exeOrder = "feat={exeDef}(df,'{exeCols}','{outputCol}')".format(exeCols=exeColumns,exeDef=exeDefFunction,outputCol=exeDefFunction+'_'+exeColumns)print("exeOrder : "+exeOrder)exec(exeOrder)return feat

4.1 TFIDF

權重計算方法經常會和余弦相似度(cosine similarity)一同使用于向量空間模型中，用以判斷兩份文件之間的相似性。當前，真正在搜索引擎等實際應用中廣泛使用的是Tf-idf 模型。Tf-idf 模型的主要思想是：如果詞w在一篇文檔d中出現的頻率高，并且在其他文檔中很少出現，則認為詞w具有很好的區分能力，適合用來把文章d和其他文章區分開來。

def TFIDF(df,inputCol="sentence",outputCol="tfidf", numFeatures=20):"""詞頻－逆向文件頻率（TF-IDF）是一種在文本挖掘中廣泛使用的特征向量化方法，它可以體現一個文檔中詞語在語料庫中的重要程度。# 總結：一個詞語在一篇文章中出現次數越多, 同時在所有文檔中出現次數越少, 越能夠代表該文章."""from pyspark.ml.feature import HashingTF, IDF, TokenizertokenizerX = Tokenizer(inputCol=inputCol, outputCol="words")wordsDataX = tokenizerX.transform(df)hashingTFX = HashingTF(inputCol="words", outputCol="rawFeatures", numFeatures=numFeatures)featurizedData = hashingTFX.transform(wordsDataX)idfX = IDF(inputCol="rawFeatures", outputCol=outputCol)idfModel = idfX.fit(featurizedData)tfidfRes = idfModel.transform(featurizedData).drop('words','rawFeatures')return tfidfRes

上述代碼輸出結果如下：

TFIDF Output +-----+------------------------------------+----------------------------------------------------------------------------------------+ |id |sentence |tfidf | +-----+------------------------------------+----------------------------------------------------------------------------------------+ |zhu |Hi I heard about pySpark |(20,[0,9,17],[0.6931471805599453,0.5753641449035617,1.3862943611198906]) | |xiang|I wish python could use case classes|(20,[2,9,13,15],[0.6931471853,1.15072071234,0.285178085,0.28768207245178085]) | |yu |Logistic regression models are neat |(20,[4,6,13,15,18],[0.69314718053,0.693147453,0.287682078085,0.2876085,0.693149453]) | +-----+------------------------------------+----------------------------------------------------------------------------------------+

4.2 Word2Vec

word2vec模型其實就是簡單化的神經網絡，它可以將文本數據向量化。詞向量具有良好的語義特性，是表示詞語特征的常用方式。詞向量每一維的值代表一個具有一定的語義和語法上解釋的特征。所以，可以將詞向量的每一維稱為一個詞語特征。詞向量具有多種形式，distributed representation 是其中一種。一個 distributed representation 是一個稠密、低維的實值向量。distributed representation 的每一維表示詞語的一個潛在特征，該特 征捕獲了有用的句法和語義特性。可見，distributed representation 中的 distributed 一詞體現了詞向量這樣一個特點：將詞語的不同句法和語義特征分布到它的每一個維度去表示。

def Word2Vec(df,inputCol="sentence",outputCol="w2v",vectorSize=100, minCount=5, numPartitions=1,stepSize=0.025, maxIter=1, seed=None, windowSize=5, maxSentenceLength=1000):"""Word2vec：將word轉化為vector，word是順序有意義的實體，比如文檔中單詞、用戶依次點擊的商品。Word2vec 得到實體向量，可以用來度量實體間相似度，在此基礎上，以下方向都可以應用：分類,聚類,推薦,句子向量,短文本分類。## 兩種實現方式# Skip-gram：用一個詞語作為輸入，來預測它周圍的上下文。# CBOW ：用一個詞語的上下文作為輸入，來預測這個詞語本身。#Spark 的 Word2vec 是一個Estimator，它采用一系列代表文檔的詞語來訓練word2vecmodel。[Spark實現的是Skip-gram模型]該模型將每個詞語映射到一個固定大小的向量。word2vecmodel使用文檔中每個詞語的平均數來將文檔轉換為向量，然后這個向量可以作為預測的特征，來計算文檔相似度計算等等。"""from pyspark.ml.feature import Word2Vecfrom pyspark.sql.functions import split# Input data: Each row is a bag of words from a sentence or document.df = df.withColumn("words",split(df[inputCol],' '))word2VecX = Word2Vec(vectorSize = vectorSize,minCount = minCount,inputCol = "words",outputCol = outputCol,numPartitions = numPartitions,stepSize = stepSize,maxIter = maxIter,seed = seed,windowSize = windowSize,maxSentenceLength = maxSentenceLength)w2vModel = word2VecX.fit(df)w2vRes = w2vModel.transform(df).drop('words')return w2vRes

上述代碼輸出結果如下：

Word2Vec Output +-----+--------------------+--------------------+ | id| sentence| w2v| +-----+--------------------+--------------------+ | zhu|Hi I heard about ...|[0.08936496693640...| |xiang|I wish python cou...|[7.36715538161141...| | yu|Logistic regressi...|[-0.0063562680035...| +-----+--------------------+--------------------+

4.3 CountVectorizer

Countvectorizer旨在通過計數來將一個文檔轉換為向量。

def CountVectorizer(df,inputCol="sentence",outputCol="cv",vectorSize=200000, minCount=1.0):"""Countvectorizer旨在通過計數來將一個文檔轉換為向量。當不存在先驗字典時，Countvectorizer可作為Estimator來提取詞匯，并生成一個Countvectorizermodel。該模型產生文檔關于詞語的稀疏表示，其表示可以傳遞給其他算法如LDA。#在fitting過程中，countvectorizer將根據語料庫中的詞頻排序選出前vocabsize個詞。一個可選的參數minDF也影響fitting過程中，它指定詞匯表中的詞語在文檔中最少出現的次數。另一個可選的二值參數控制輸出向量，如果設置為真那么所有非零的計數為1。這對于二值型離散概率模型非常有用。"""from pyspark.ml.feature import CountVectorizerfrom pyspark.sql.functions import splitdf = df.withColumn("words",split(df[inputCol],' '))CountVectorizerX = CountVectorizer(inputCol="words", outputCol=outputCol, vocabSize=vectorSize, minDF=minCount)cvModelX = CountVectorizerX.fit(df)cvRes = cvModelX.transform(df).drop('words')return cvRes

上述代碼輸出結果如下：

CountVectorizer Output +-----+------------------------------------+----------------------------------------------------+ |id |sentence |cv | +-----+------------------------------------+----------------------------------------------------+ |zhu |Hi I heard about pySpark |(16,[0,2,4,12,13],[1.0,1.0,1.0,1.0,1.0]) | |xiang|I wish python could use case classes|(16,[0,3,5,6,8,10,14],[1.0,1.0,1.0,1.0,1.0,1.0,1.0])| |yu |Logistic regression models are neat |(16,[1,7,9,11,15],[1.0,1.0,1.0,1.0,1.0]) | +-----+------------------------------------+----------------------------------------------------+

4.4 OneHotEncoder

將類別特征映射為二進制向量，其中只有一個有效值(為1，其余為0)。

def OneHotEncoder(df,inputCol="category",outputCol="categoryVec"):"""將類別特征映射為二進制向量，其中只有一個有效值(為1，其余為0)。"""from pyspark.ml.feature import OneHotEncoder, StringIndexerstringIndexerX = StringIndexer(inputCol=inputCol, outputCol="categoryIndex")modelX = stringIndexerX.fit(df)indexed = modelX.transform(df)encoderX = OneHotEncoder(inputCol="categoryIndex", outputCol=outputCol)encodedX = encoderX.transform(indexed).drop("categoryIndex")return encodedX

上述代碼輸出結果如下：

OneHotEncoder Output +-------+--------+-------------+-------------+ | id|category|categoryIndex| categoryVec| +-------+--------+-------------+-------------+ | zhu| a| 0.0|(2,[0],[1.0])| |xiangyu| b| 2.0| (2,[],[])| | yu| c| 1.0|(2,[1],[1.0])| | is| a| 0.0|(2,[0],[1.0])| | coming| a| 0.0|(2,[0],[1.0])| | now| c| 1.0|(2,[1],[1.0])| +-------+--------+-------------+-------------+

4.5 StringIndexer

將標簽索引化，然后索引數值根據標簽出現的頻率進行排序。

def StringIndexer(df,inputCol="category",outputCol="categoryVec"):"""將標簽索引化，然后索引數值根據標簽出現的頻率進行排序。"""from pyspark.ml.feature import StringIndexerindexerX = StringIndexer(inputCol=inputCol, outputCol=outputCol)indexedX = indexerX.fit(df).transform(df)return indexedX

上述代碼輸出結果如下：

StringIndexer Output +-----+--------------------+-----------+ | id| sentence|categoryVec| +-----+--------------------+-----------+ | zhu|Hi I heard about ...| 2.0| |xiang|I wish python cou...| 0.0| | yu|Logistic regressi...| 1.0| +-----+--------------------+-----------+

4.6 IndexToString

與StringIndexer對應，IndexToString將索引化標簽還原成原始字符串。

def IndexToString(df,inputCol="categoryVec",outputCol="category"):"""與StringIndexer對應，IndexToString將索引化標簽還原成原始字符串。"""from pyspark.ml.feature import IndexToStringconverterX = IndexToString(inputCol=inputCol, outputCol=outputCol)convertedX = converterX.transform(df)return convertedX

上述代碼輸出結果如下：

IndexToString Output IndexToString(StringIndexer(df,"sentence")) +-----+--------------------+-----------+--------------------+ | id| sentence|categoryVec| category| +-----+--------------------+-----------+--------------------+ | zhu|Hi I heard about ...| 2.0|Hi I heard about ...| |xiang|I wish python cou...| 0.0|I wish python cou...| | yu|Logistic regressi...| 1.0|Logistic regressi...| +-----+--------------------+-----------+--------------------+

4.7 PCA

主成分分析是一種對數據進行旋轉變換的統計學方法，其本質是在線性空間中進行一個基變換，使得變換后的數據投影在一組新的"坐標軸"上的方差最大化，隨后，裁剪掉變換后方差很小的"坐標軸"，剩下的新的"坐標軸"即被稱為主成分，它們可以再一個較低維度的子空間中盡可能地表示原有數據的性質。

PCA Input +--------------------+ | features| +--------------------+ | (5,[1,3],[1.0,7.0])| |[2.0,0.0,3.0,4.0,...| |[4.0,0.0,0.0,6.0,...| +--------------------+

上述代碼輸出結果如下：

def PCA(df,vectorSize=3, inputCol="features", outputCol="pcaFeatures"):"""主成分分析是一種對數據進行旋轉變換的統計學方法，其本質是在線性空間中進行一個基變換，使得變換后的數據投影在一組新的"坐標軸"上的方差最大化，隨后，裁剪掉變換后方差很小的"坐標軸"，剩下的新的"坐標軸"即被稱為主成分，它們可以再一個較低維度的子空間中盡可能地表示原有數據的性質。"""from pyspark.ml.feature import PCApcaX = PCA(k=vectorSize, inputCol=inputCol, outputCol=outputCol)modelX = pcaX.fit(df)pcaRes = modelX.transform(df)return pcaRes

上述代碼輸出結果如下：

PCA Output +-----------------------------------------------------------+ |pcaFeatures | +-----------------------------------------------------------+ |[1.6485728230883807,-4.013282700516296,-5.524543751369388] | |[-4.645104331781534,-1.1167972663619026,-5.524543751369387]| |[-6.428880535676489,-5.337951427775355,-5.524543751369389] | +-----------------------------------------------------------+

4.8 Binarizer

把數值型特征值轉化成二進制(0/1)輸出，設置一個閾值，大于閾值的輸出1，小于閾值的輸出0。

def Binarizer(df,threshold=0.5, inputCol="feature", outputCol="binarized_feature"):"""把數值型特征值轉化成二進制(0/1)輸出，設置一個閾值，大于閾值的輸出1，小于閾值的輸出0"""from pyspark.ml.feature import BinarizerbinarizerX = Binarizer(threshold=threshold, inputCol=inputCol, outputCol=outputCol)binarizedX = binarizerX.transform(df)return binarizedX

上述代碼輸出結果如下：

Binarizer Output +---+-------+-----------------+ | id|feature|binarized_feature| +---+-------+-----------------+ | 0| 0.1| 0.0| | 1| 0.8| 1.0| | 2| 0.2| 0.0| +---+-------+-----------------+

4.9 Tokenizer

分詞器:提供默認分詞，也提供正則表達式分詞

def Tokenizer(df,inputCol="sentence", outputCol="words", pattern="\\W"):"""分詞器:提供默認分詞，也提供正則表達式分詞"""from pyspark.ml.feature import RegexTokenizerregexTokenizer = RegexTokenizer(inputCol=inputCol, outputCol=outputCol, pattern=pattern)regexTokenized = regexTokenizer.transform(df)return regexTokenized

上述代碼輸出結果如下：

Tokenizer Output +-----------------------------------+------------------------------------------+------+ |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.10 StopWordsRemover

停用詞過濾

def StopWordsRemover(df,inputCol="words", outputCol="words2",add_stopwords=[]):"""停用詞過濾"""from pyspark.ml.feature import StopWordsRemoverremover = StopWordsRemover(inputCol=inputCol, outputCol=outputCol).setStopWords(add_stopwords)# 添加停用詞# remover = remover.setStopWords(Array("saw","Mary"))removed = remover.transform(df)return removed

上述代碼輸出結果如下：

StopWordsRemover Output +---+----------------------------+--------------------+ |id |words |words2 | +---+----------------------------+--------------------+ |0 |[I, saw, the, red, balloon] |[saw, red, balloon] | |1 |[Mary, had, a, little, lamb]|[Mary, little, lamb]| +---+----------------------------+--------------------+

4.11 NGram

把單詞轉成一個個連續詞輸出。

def NGram(df,n=2, inputCol="words", outputCol="ngrams"):"""把單詞轉成一個個連續詞輸出"""from pyspark.ml.feature import NGramngram = NGram(n=2, inputCol=inputCol, outputCol=outputCol)ngramDF = ngram.transform(df)return ngramDF

上述代碼輸出結果如下：

NGram Output +---+--------------------------------------------+----------------------------------------------------------------------+ |id |words |ngrams | +---+--------------------------------------------+----------------------------------------------------------------------+ |0 |[Hi, I, heard, about, Spark] |[Hi I, I heard, heard about, about Spark] | |1 |[I, wish, python, could, use, case, classes]|[I wish, wish python, python could, could use, use case, case classes]| |2 |[Logistic, regression, models, are, neat] |[Logistic regression, regression models, models are, are neat] | +---+--------------------------------------------+----------------------------------------------------------------------+

4.12 DCT

離散余弦變換是將時域的N維實數序列轉換成頻域的N維實數序列的過程(有點類似離散傅里葉變換)。

def DCT(df, inverse=False, inputCol="features", outputCol="featuresDCT"):"""離散余弦變換是將時域的N維實數序列轉換成頻域的N維實數序列的過程(有點類似離散傅里葉變換)。"""from pyspark.ml.feature import DCTdct = DCT(inverse=inverse, inputCol=inputCol, outputCol=outputCol)dctDf = dct.transform(df)return dctDf

4.13 ChiSqSelector

ChiSqSelector代表卡方特征選擇。ChiSqSelector根據獨立卡方檢驗，然后選取類別標簽主要依賴的特征。 selectorType Supported options: numTopFeatures (default), percentile and fpr. - 1、numTopFeatures：通過卡方檢驗選取最具有預測能力的Top(num)個特征 - 2、percentile：類似于上一種方法，但是選取一小部分特征而不是固定(num)個特征 - 3、fpr:選擇P值低于門限值的特征，這樣就可以控制false positive rate來進行特征選擇

def ChiSqSelector(df, featuresCol='features', labelCol='label',numTopFeatures=50,outputCol="selectedFeatures",selectorType='numTopFeatures', percentile=0.1, fpr=0.05):"""ChiSqSelector代表卡方特征選擇。ChiSqSelector根據獨立卡方檢驗，然后選取類別標簽主要依賴的特征。"""# selectorType Supported options: numTopFeatures (default), percentile and fpr.# 1、numTopFeatures：通過卡方檢驗選取最具有預測能力的Top(num)個特征# 2、percentile：類似于上一種方法，但是選取一小部分特征而不是固定(num)個特征# 3、fpr:選擇P值低于門限值的特征，這樣就可以控制false positive rate來進行特征選擇from pyspark.ml.feature import ChiSqSelectorselector = ChiSqSelector(numTopFeatures = numTopFeatures,featuresCol = featuresCol,outputCol = outputCol,labelCol = labelCol,selectorType = selectorType,percentile = percentile,fpr = fpr)result = selector.fit(df).transform(df)print("ChiSqSelector output with top %d features selected" % selector.getNumTopFeatures())return result

上述代碼輸出結果如下：

# ChiSqSelector Output with top 1 features selected +---+------------------+-------+----------------+ | id| features| label|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.14 PearsonCorr

皮爾遜相關系數( Pearson correlation coefficient） 用于度量兩個變量X和Y之間的相關（線性相關），其值介于-1與1之間。

def PearsonCorr(df,featureCol='feature',labelCol='label'):"""皮爾遜相關系數( Pearson correlation coefficient）用于度量兩個變量X和Y之間的相關（線性相關），其值介于-1與1之間。"""return df.corr(featureCol,labelCol,method=None)

總結

以上是生活随笔為你收集整理的推荐算法工程笔记：PySpark特征工程入门总结的全部內容，希望文章能夠幫你解決所遇到的問題。

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