spark mllib 特征学习笔记 （一）

PySpark MLlib 特征处理详解

PySpark MLlib 提供了丰富的特征处理工具，帮助我们进行特征提取、转换和选择。以下是 PySpark MLlib 中常用的特征处理类及其简要介绍。

1. Binarizer

Binarizer 是将连续特征二值化的转换器。

from pyspark.ml.feature import Binarizerbinarizer = Binarizer(threshold=0.5, inputCol="feature", outputCol="binarized_feature")
binarizedData = binarizer.transform(data)

2. BucketedRandomProjectionLSH

BucketedRandomProjectionLSH 是基于欧几里得距离度量的 LSH 类。

from pyspark.ml.feature import BucketedRandomProjectionLSHbrp = BucketedRandomProjectionLSH(inputCol="features", outputCol="hashes", bucketLength=2.0)
model = brp.fit(data)
transformedData = model.transform(data)

3. Bucketizer

Bucketizer 将连续特征映射到特征桶。

from pyspark.ml.feature import Bucketizersplits = [-float("inf"), 0.0, float("inf")]
bucketizer = Bucketizer(splits=splits, inputCol="feature", outputCol="bucketed_feature")
bucketedData = bucketizer.transform(data)

4. ChiSqSelector

ChiSqSelector 是卡方特征选择器，选择预测分类标签的分类特征。

from pyspark.ml.feature import ChiSqSelectorselector = ChiSqSelector(numTopFeatures=50, featuresCol="features", labelCol="label", outputCol="selected_features")
result = selector.fit(data).transform(data)

5. CountVectorizer

CountVectorizer 从文档集合中提取词汇，并生成 CountVectorizerModel。

from pyspark.ml.feature import CountVectorizercv = CountVectorizer(inputCol="text", outputCol="features", vocabSize=10000, minDF=5)
model = cv.fit(data)
vectorizedData = model.transform(data)

6. DCT

DCT 是对实数向量进行一维离散余弦变换的特征转换器。

from pyspark.ml.feature import DCTdct = DCT(inverse=False, inputCol="features", outputCol="dct_features")
dctData = dct.transform(data)

7. ElementwiseProduct

ElementwiseProduct 对每个输入向量与提供的“权重”向量进行 Hadamard 乘积（即逐元素乘积）。

from pyspark.ml.feature import ElementwiseProduct
from pyspark.ml.linalg import VectorsscalingVec = Vectors.dense([0.0, 1.0, 2.0])
transformer = ElementwiseProduct(scalingVec=scalingVec, inputCol="features", outputCol="scaled_features")
scaledData = transformer.transform(data)

8. FeatureHasher

FeatureHasher 将一组分类或数值特征投影到指定维度的特征向量中。

from pyspark.ml.feature import FeatureHasherhasher = FeatureHasher(inputCols=["cat1", "cat2", "num1"], outputCol="features")
hashedData = hasher.transform(data)

9. HashingTF

HashingTF 使用哈希技巧将词序列映射到它们的词频。

from pyspark.ml.feature import HashingTFhashingTF = HashingTF(inputCol="text", outputCol="features", numFeatures=10000)
tfData = hashingTF.transform(data)

10. IDF

IDF 计算文档集合的逆文档频率（IDF）。

from pyspark.ml.feature import IDFidf = IDF(inputCol="raw_features", outputCol="features", minDocFreq=5)
model = idf.fit(tfData)
tfidfData = model.transform(tfData)

11. Imputer

Imputer 使用列中的均值、中位数或众数来填补缺失值。

from pyspark.ml.feature import Imputerimputer = Imputer(inputCols=["feature1", "feature2"], outputCols=["imputed_feature1", "imputed_feature2"])
model = imputer.fit(data)
imputedData = model.transform(data)

12. IndexToString

IndexToString 将索引列映射回相应的字符串值列。

from pyspark.ml.feature import IndexToStringconverter = IndexToString(inputCol="index", outputCol="string", labels=["a", "b", "c"])
convertedData = converter.transform(data)

13. Interaction

Interaction 实现特征交互转换。

from pyspark.ml.feature import Interactioninteraction = Interaction(inputCols=["col1", "col2"], outputCol="interacted_col")
interactedData = interaction.transform(data)

14. MaxAbsScaler

MaxAbsScaler 通过除以每个特征的最大绝对值来单独缩放每个特征到范围 [-1, 1]。

from pyspark.ml.feature import MaxAbsScalerscaler = MaxAbsScaler(inputCol="features", outputCol="scaled_features")
model = scaler.fit(data)
scaledData = model.transform(data)

15. MinHashLSH

MinHashLSH 是基于 Jaccard 距离的 LSH 类。

from pyspark.ml.feature import MinHashLSHmh = MinHashLSH(inputCol="features", outputCol="hashes", numHashTables=3)
model = mh.fit(data)
transformedData = model.transform(data)

16. MinMaxScaler

MinMaxScaler 使用列摘要统计数据，将每个特征单独线性缩放到 [min, max] 范围内，也称为最小-最大归一化或重缩放。

from pyspark.ml.feature import MinMaxScalerscaler = MinMaxScaler(inputCol="features", outputCol="scaled_features")
model = scaler.fit(data)
scaledData = model.transform(data)

17. NGram

NGram 是一个特征转换器，它将输入的字符串数组转换为 n-grams 数组。

from pyspark.ml.feature import NGramngram = NGram(n=2, inputCol="words", outputCol="ngrams")
ngramData = ngram.transform(data)

18. Normalizer

Normalizer 使用给定的 p-范数将向量规范化为单位范数。

from pyspark.ml.feature import Normalizernormalizer = Normalizer(p=1.0, inputCol="features", outputCol="norm_features")
normData = normalizer.transform(data)

19. OneHotEncoder

OneHotEncoder 将分类索引列映射到二进制向量列。

from pyspark.ml.feature import OneHotEncoderencoder = OneHotEncoder(inputCol="index", outputCol="onehot")
encodedData = encoder.transform(data)

20. PCA

PCA 训练一个模型，将向量投影到前 k 个主成分的低维空间中。

from pyspark.ml.feature import PCApca = PCA(k=3, inputCol="features", outputCol="pca_features")
model = pca.fit(data)
pcaData = model.transform(data)

21. PolynomialExpansion

PolynomialExpansion 在多项式空间中进行特征扩展。

from pyspark.ml.feature import PolynomialExpansionpolyExpansion = PolynomialExpansion(degree=2, inputCol="features", outputCol="poly_features")
polyData = polyExpansion.transform(data)

22. QuantileDiscretizer

QuantileDiscretizer 将连续特征列离散化为分类特征列。

from pyspark.ml.feature import QuantileDiscretizerdiscretizer = QuantileDiscretizer(numBuckets=3, inputCol="feature", outputCol="bucketed_feature")
bucketedData = discretizer.fit(data).transform(data)

23. RobustScaler

RobustScaler 移除中位数并根据四分位范围缩放数据。

from pyspark.ml.feature import RobustScalerscaler = RobustScaler(inputCol="features", outputCol="scaled_features")
model = scaler.fit(data)
scaledData = model.transform(data)

24. RegexTokenizer

RegexTokenizer 是一个基于正则表达式的分词器，可以使用提供的正则表达式模式（默认为分隔模式）提取标记，或反复匹配正则表达式（如果 gaps 为 false）。

from pyspark.ml.feature import RegexTokenizertokenizer = RegexTokenizer(inputCol="text", outputCol="words", pattern="\\W")
tokenizedData = tokenizer.transform(data)

25. RFormula

RFormula 实现了对数据集进行拟合所需的转换，使用 R 模型公式。

from pyspark.ml.feature import RFormulaformula = RFormula(formula="y ~ x1 + x2", featuresCol="features", labelCol="label")
formulaData = formula.fit(data).transform(data)

26. SQLTransformer

SQLTransformer 实现了由 SQL 语句定义的转换。

from pyspark.ml.feature import SQLTransformersqlTrans = SQLTransformer(statement="SELECT *, (col1 + col2) AS new_col FROM __THIS__")
transformedData = sqlTrans.transform(data)

27. StandardScaler

StandardScaler 使用训练集中的样本列摘要统计数据，通过去均值和按单位方差缩放来标准化特征。

from pyspark.ml.feature import StandardScalerscaler = StandardScaler(inputCol="features", outputCol="scaled_features", withMean=True, withStd=True)
model = scaler.fit(data)
scaledData = model.transform(data)

28. StopWordsRemover

StopWordsRemover 是一个特征转换器，用于从输入中过滤停用词。

from pyspark.ml.feature import StopWordsRemoverremover = StopWordsRemover(inputCol="raw", outputCol="filtered")
filteredData = remover.transform(data)

29. StringIndexer

StringIndexer 是一个标签索引器，将字符串标签列映射到标签索引列。

from pyspark.ml.feature import StringIndexerindexer = StringIndexer(inputCol="category", outputCol="categoryIndex")
indexedData = indexer.fit(data).transform(data)

30. Tokenizer

Tokenizer 是一个分词器，将输入字符串转换为小写，然后按空格拆分。

from pyspark.ml.feature import Tokenizertokenizer = Tokenizer(inputCol="text", outputCol="words")
tokenizedData = tokenizer.transform(data)

31. UnivariateFeatureSelector

UnivariateFeatureSelector 基于单变量统计测试选择特征。

from pyspark.ml.feature import UnivariateFeatureSelectorselector = UnivariateFeatureSelector(featuresCol="features", labelCol="label", selectionMode="numTopFeatures", selectionThreshold=50)
selectedData = selector.fit(data).transform(data)

32. VarianceThresholdSelector

VarianceThresholdSelector 删除所有低方差特征的特征选择器。

from pyspark.ml.feature import VarianceThresholdSelectorselector = VarianceThresholdSelector(featuresCol="features", varianceThreshold=0.1, outputCol="selected_features")
selectedData = selector.fit(data).transform(data)

33. VectorAssembler

VectorAssembler 是一个特征转换器，将多个列合并为一个向量列。

from pyspark.ml.feature import VectorAssemblerassembler = VectorAssembler(inputCols=["col1", "col2", "col3"], outputCol="features")
assembledData = assembler.transform(data)

34. VectorIndexer

VectorIndexer 是用于对数据集中 Vector 的分类特征列进行索引的类。

from pyspark.ml.feature import VectorIndexerindexer = VectorIndexer(inputCol="features", outputCol="indexed_features", maxCategories=10)
indexerModel = indexer.fit(data)
indexedData = indexerModel.transform(data)

35. VectorSizeHint

VectorSizeHint 是一个特征转换器，向向量列的元数据添加大小信息。

from pyspark.ml.feature import VectorSizeHintsizeHint = VectorSizeHint(inputCol="features", size=3)
hintedData = sizeHint.transform(data)

36. VectorSlicer

VectorSlicer 是一个类，接收一个特征向量，并输出一个新的特征向量，其中包含原始特征的子数组。

from pyspark.ml.feature import VectorSlicerslicer = VectorSlicer(inputCol="features", outputCol="sliced_features", indices=[1, 2])
slicedData = slicer.transform(data)

37. Word2Vec

Word2Vec 训练一个 Map(String, Vector) 的模型，即将字词映射到向量。

from pyspark.ml.feature import Word2Vecword2Vec = Word2Vec(inputCol="text", outputCol="result", vectorSize=3, minCount=0)
model = word2Vec.fit(data)
resultData = model.transform(data)

以下是 PySpark MLlib 中部分特征处理方法的详细介绍，包括它们所基于的公式、适用的场景以及一些具体的应用案例。

1. Binarizer

公式：
$$\text{Binarizer}(x)=\{\begin{array}{ll}1& \text{if }x>\text{threshold}\\ 0& \text{otherwise}\end{array}$$

适用场景：
用于将连续特征转换为二值特征，常用于分类问题中将数值特征转换为二进制特征。

案例：

from pyspark.ml.feature import Binarizerdata = spark.createDataFrame([(0.1,), (0.8,), (0.5,)], ["feature"])
binarizer = Binarizer(threshold=0.5, inputCol="feature", outputCol="binarized_feature")
binarizedData = binarizer.transform(data)
binarizedData.show()

2. Bucketizer

公式：
将连续特征分成离散的桶，例如使用指定的分割点将特征值分段：
$$\text{Bucketizer}(x)=\{\begin{array}{ll}0& \text{if }x\le \text{splits}[1]\\ 1& \text{if }\text{splits}[1]<x\le \text{splits}[2]\\ ⋮& ⋮\\ N-1& \text{if }x>\text{splits}[N-1]\end{array}$$

适用场景：
用于将连续特征转换为离散的分桶特征，常用于决策树等算法中。

案例：

from pyspark.ml.feature import Bucketizerdata = spark.createDataFrame([(0.1,), (0.8,), (0.5,)], ["feature"])
splits = [-float("inf"), 0.5, float("inf")]
bucketizer = Bucketizer(splits=splits, inputCol="feature", outputCol="bucketed_feature")
bucketedData = bucketizer.transform(data)
bucketedData.show()

3. ChiSqSelector

公式：
根据卡方检验的统计量选择特征：
$${\chi }^2=\sum \frac{(O_i-E_i{)}^2}{E_i}$$
其中 (O_i) 是观察频数，(E_i) 是期望频数。

适用场景：
用于特征选择，特别是用于分类问题中的分类特征选择。

案例：

from pyspark.ml.feature import ChiSqSelector
from pyspark.ml.linalg import Vectorsdata = spark.createDataFrame([(Vectors.dense([0.0, 0.5, 0.5]), 1.0),(Vectors.dense([0.1, 0.8, 0.2]), 0.0),(Vectors.dense([0.2, 0.9, 0.1]), 0.0)
], ["features", "label"])selector = ChiSqSelector(numTopFeatures=2, featuresCol="features", labelCol="label", outputCol="selected_features")
result = selector.fit(data).transform(data)
result.show()

4. CountVectorizer

公式：
计算词汇表并生成词频向量：
$$\text{CountVectorizer}(D)=[\text{TF}(t_1,D),\text{TF}(t_2,D),\dots ,\text{TF}(t_n,D)]$$
其中 (\text{TF}(t_i, D)) 是词 (t_i) 在文档 (D) 中的词频。

适用场景：
用于文本数据的词频特征提取，常用于自然语言处理和文本分类任务。

案例：

from pyspark.ml.feature import CountVectorizerdata = spark.createDataFrame([(0, "a b c".split(" ")), (1, "a b b c a".split(" "))], ["id", "words"])
cv = CountVectorizer(inputCol="words", outputCol="features", vocabSize=3, minDF=1)
model = cv.fit(data)
vectorizedData = model.transform(data)
vectorizedData.show()

5. DCT

公式：
离散余弦变换 (DCT)：
$$X_k=\sum _{n=0}^{N-1}{x}_n\cos \left[\frac{\pi }{N}\left(n+\frac{1}{2}\right)k\right]$$

适用场景：
用于信号处理中的特征转换，如图像处理和压缩。

案例：

from pyspark.ml.feature import DCT
from pyspark.ml.linalg import Vectorsdata = spark.createDataFrame([(Vectors.dense([0.0, 1.0, -2.0, 3.0]),)], ["features"])
dct = DCT(inverse=False, inputCol="features", outputCol="dct_features")
dctData = dct.transform(data)
dctData.show()

6. Imputer

公式：
缺失值填充，使用均值、中位数或众数填充：
$$\text{Imputer}(x)=\{\begin{array}{ll}x& \text{if }x\ne \text{NaN}\\ \text{mean/median/mode}(X)& \text{if }x=\text{NaN}\end{array}$$

适用场景：
用于处理数据集中的缺失值。

案例：

from pyspark.ml.feature import Imputerdata = spark.createDataFrame([(1.0, float("nan")), (2.0, 3.0), (float("nan"), 4.0)], ["a", "b"])
imputer = Imputer(inputCols=["a", "b"], outputCols=["imputed_a", "imputed_b"])
model = imputer.fit(data)
imputedData = model.transform(data)
imputedData.show()

7. OneHotEncoder

公式：
将分类特征转换为独热编码向量：
$$\text{OneHotEncoder}(x)=[0,\dots ,1,\dots ,0]$$
其中1出现在类别索引位置。

适用场景：
用于将分类特征转换为机器学习算法可以直接使用的数值特征。

案例：

from pyspark.ml.feature import OneHotEncoder, StringIndexerdata = spark.createDataFrame([("a",), ("b",), ("a",)], ["category"])
indexer = StringIndexer(inputCol="category", outputCol="categoryIndex")
indexed = indexer.fit(data).transform(data)
encoder = OneHotEncoder(inputCol="categoryIndex", outputCol="categoryVec")
encoded = encoder.fit(indexed).transform(indexed)
encoded.show()

8. PCA

公式：
主成分分析 (PCA)：
$$\mathbf{X}=\mathbf{T}{\mathbf{P}}^T$$
其中 (\mathbf{T}) 是得分矩阵，(\mathbf{P}) 是载荷矩阵。

适用场景：
用于降维，提取主要特征，减少数据集的维度。

案例：

from pyspark.ml.feature import PCA
from pyspark.ml.linalg import Vectorsdata = spark.createDataFrame([(Vectors.dense([1.0, 0.0, 0.0]),), (Vectors.dense([0.0, 1.0, 0.0]),), (Vectors.dense([0.0, 0.0, 1.0]),)], ["features"])
pca = PCA(k=2, inputCol="features", outputCol="pca_features")
model = pca.fit(data)
pcaData = model.transform(data)
pcaData.show()

9. StandardScaler

公式：
标准化特征，去均值并按标准差缩放：
$$\text{StandardScaler}(x)=\frac{x-\text{mean}(x)}{\text{std}(x)}$$

适用场景：
用于特征标准化，使不同特征具有相同的尺度，适用于大多数机器学习算法。

案例：

from pyspark.ml.feature import StandardScaler
from pyspark.ml.linalg import Vectorsdata = spark.createDataFrame([(Vectors.dense([1.0, 0.1, -1.0]),), (Vectors.dense([2.0, 1.1, 1.0]),), (Vectors.dense([4.0, 10.1, 2.0]),)], ["features"])
scaler = StandardScaler(inputCol="features", outputCol="scaled_features", withMean=True, withStd=True)
model = scaler.fit(data)
scaledData = model.transform(data)
scaledData.show()
``
## 总结PySpark MLlib 中的特征处理工具丰富且功能强大，可以帮助我们在数据预处理阶段完成各种特征工程任务。这些工具覆盖了特征的二值化、离散化、标准化、归一化、编码、转换、选择和生成等多个方面，是数据科学家和工程师进行机器学习模型训练的重要帮手。通过合理使用这些工具，可以极大提升模型的性能和效果。