ReaderMX.
数据STUDIO

PySpark ML 特征转换方法总结 (上)

数云朵君约 12 分钟

日前云朵君跟大家一起学习了 (可戳👉) PySpark ML 特征提取与选择方法总结 ，今天给大家总结了12个数据挖掘特征转换方法简介及代码示例，建议收藏！

  • Tokenizer

  • StopWordsRemover

  • N-gram

  • Binarizer

  • PCA

  • PolynomialExpansion

  • 离散余弦变换(DCT)

  • StringIndexer

  • IndexToString

  • OneHotEncoder

  • VectorIndexer

  • Interaction

Tokenizer

标记化Tokenizer是获取文本（例如句子）并将其分解为单个术语（通常是单词）的过程。

正则化的标记化RegexTokenizer基于正则表达式 (regex) 匹配的更高级的标记化。默认情况下，参数“pattern”（正则表达式，默认值： "\\s+" ）用作分隔输入文本的分隔符。可以将参数“gaps”设置为 false，表示正则表达式“pattern”表示“tokens”而不是拆分gaps，并找到所有匹配的出现作为标记化结果。 

from pyspark.ml.feature import Tokenizer, RegexTokenizerfrom pyspark.sql.functions import col, udffrom pyspark.sql.types import IntegerTypesentenceDataFrame = spark.createDataFrame([      (0, "Hi I heard about PyDataStudio"),      (1, "I wish you will follow us"),      (2, "Logistic,regression,models,are,neat")], ["id", "sentence"])tokenizer = Tokenizer(inputCol="sentence", outputCol="words")regexTokenizer = RegexTokenizer(inputCol="sentence", outputCol="words", pattern="\\W")# alternatively, pattern="\\w+", gaps(False)countTokens = udf(lambda words: len(words), IntegerType())tokenized = tokenizer.transform(sentenceDataFrame)tokenized.select("sentence", "words")\    .withColumn("tokens", countTokens(col("words"))).show(truncate=False)regexTokenized = regexTokenizer.transform(sentenceDataFrame)regexTokenized.select("sentence", "words") \    .withColumn("tokens", countTokens(col("words"))).show(truncate=False)

输出： 

+-----------------------------------+-------------------------------------+------+|sentence                           |words                                |tokens|+-----------------------------------+-------------------------------------+------+|Hi I heard about PyDataStudio      |[hi, i, heard, about, pydatastudio]  |5     ||I wish you will follow us          |[i, wish, you, will, follow, us]     |6     ||Logistic,regression,models,are,neat|[logistic,regression,models,are,neat]|1     |+-----------------------------------+-------------------------------------+------++-----------------------------------+-----------------------------------------+------+|sentence                           |words                                    |tokens|+-----------------------------------+-----------------------------------------+------+|Hi I heard about PyDataStudio      |[hi, i, heard, about, pydatastudio]      |5     ||I wish you will follow us          |[i, wish, you, will, follow, us]         |6     ||Logistic,regression,models,are,neat|[logistic, regression, models, are, neat]|5     |+-----------------------------------+-----------------------------------------+------+

StopWordsRemover

停用词去除器是应该从输入中排除的词，通常是因为这些词经常出现并且没有太多意义。

StopWordsRemover 将字符串序列（例如Tokenizer的输出）作为输入，并从输入序列中删除所有停用词。停用词列表由 stopWords 参数指定。 

from pyspark.ml.feature import StopWordsRemoversentenceData = spark.createDataFrame([    (0, ["I", "saw", "the", "red", "balloon"]),    (1, ["Mary", "had", "a", "little", "lamb"]) ], ["id", "raw"])remover = StopWordsRemover(inputCol="raw",                            outputCol="filtered")remover.transform(sentenceData).show(truncate=False)

输出： 

+---+----------------------------+--------------------+|id |raw                         |filtered            |+---+----------------------------+--------------------+|0  |[I, saw, the, red, balloon] |[saw, red, balloon] ||1  |[Mary, had, a, little, lamb]|[Mary, little, lamb]|+---+----------------------------+--------------------+

N-gram

n-gram是n个标记(通常是单词)的序列，用于整数n。NGram类可以用来将输入特征转换为n-gram。

NGram以字符串序列作为输入(例如，Tokenizer的输出)。参数n用于确定每个n-gram中的项数。输出将由一个n-gram序列组成，其中每个n-gram由一个由n个连续单词组成的以空格分隔的字符串表示。如果输入序列包含少于n个字符串，则不产生输出。

截至 2020 年 5 月 7 日，在有关 2019 年冠状病毒病 (COVID-19) 的出版物标题中经常出现六个 n-gram 
from pyspark.ml.feature import NGramwordDataFrame = spark.createDataFrame([    (0, ["Hi", "I", "heard", "about", "PyDataStudio"]),    (1, ["I", "wish", "you", "will", "follow", "us"]),    (2, ["Logistic", "regression", "models", "are", "neat"])], ["id", "words"])ngram = NGram(n=2, inputCol="words", outputCol="ngrams")ngramDataFrame = ngram.transform(wordDataFrame)ngramDataFrame.select("ngrams").show(truncate=False)

输出： 

+--------------------------------------------------------------+|ngrams                                                        |+--------------------------------------------------------------+|[Hi I, I heard, heard about, about PyDataStudio]              ||[I wish, wish you, you will, will follow, follow us]          ||[Logistic regression, regression models, models are, are neat]|+--------------------------------------------------------------+

Binarizer

二值化是将数值特征阈值化为二值(0/1)特征的过程。

Binarizer接受公共参数inputCol和outputCol，以及二值化的阈值。

  • 大于阈值的特征值被二值化到1.0
  • 等于或小于阈值的值被二进制化为0.0 
from pyspark.ml.feature import BinarizercontinuousDataFrame = spark.createDataFrame([    (0, 0.1),    (1, 0.8),    (2, 0.2)], ["id", "feature"])binarizer = Binarizer(threshold=0.5,                       inputCol="feature",                        # inputCol同时支持Vector和Double类型                      outputCol="binarized_feature")binarizedDataFrame = binarizer.transform(continuousDataFrame)print("带阈值的二值化输出 = %f" % binarizer.getThreshold())binarizedDataFrame.show()

输出： 

带阈值的二值化输出 = 0.500000+---+-------+-----------------+| id|feature|binarized_feature|+---+-------+-----------------+|  0|    0.1|              0.0||  1|    0.8|              1.0||  2|    0.2|              0.0|+---+-------+-----------------+

PCA

主成分分析是一种统计过程，它使用正交变换将一组可能相关的变量的观测值转换为一组线性不相关的变量的值，称为主成分。PCA类训练模型使用PCA将向量投影到低维空间。

下面的示例展示了如何将5维特征向量投影到3维主分量。 

from pyspark.ml.feature import PCAfrom pyspark.ml.linalg import Vectorsdata = [(Vectors.sparse(5, [(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]),)]df = spark.createDataFrame(data, ["features"])pca = PCA(k=3, inputCol="features",           outputCol="pcaFeatures")model = pca.fit(df)result = model.transform(df).select("pcaFeatures")result.show(truncate=False)

输出： 

+------------------------------------------------------------+|pcaFeatures                                                 |+------------------------------------------------------------+|[1.6485728230883814,-4.0132827005162985,-1.0091435193998504]||[-4.645104331781533,-1.1167972663619048,-1.0091435193998504]||[-6.428880535676488,-5.337951427775359,-1.0091435193998508] |+------------------------------------------------------------+

PolynomialExpansion

PolynomialExpansion多项式扩展是将你的特征扩展到一个多项式空间的过程，它是由原始维数的n次组合形成的。多项式扩展类提供了这种功能。

下面的示例展示了如何将特征扩展到3次多项式空间中。 

from pyspark.ml.feature import PolynomialExpansionfrom pyspark.ml.linalg import Vectorsdf = spark.createDataFrame([    (Vectors.dense([2.0, 1.0]),),    (Vectors.dense([0.0, 0.0]),),    (Vectors.dense([3.0, -1.0]),)], ["features"])polyExpansion = PolynomialExpansion(degree=3,                                     inputCol="features",                                     outputCol="polyFeatures")polyDF = polyExpansion.transform(df)polyDF.show(truncate=False)

输出： 

+----------+------------------------------------------+|features  |polyFeatures                              |+----------+------------------------------------------+|[2.0,1.0] |[2.0,4.0,8.0,1.0,2.0,4.0,1.0,2.0,1.0]     ||[0.0,0.0] |[0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0]     ||[3.0,-1.0]|[3.0,9.0,27.0,-1.0,-3.0,-9.0,1.0,3.0,-1.0]|+----------+------------------------------------------+

离散余弦变换(DCT)

Discrete Cosine Transform离散余弦变换是在时域将一个长度为N的实值序列变换为在频域的另一个长度为N的实值序列。 

from pyspark.ml.feature import DCTfrom pyspark.ml.linalg import Vectorsdf = spark.createDataFrame([    (Vectors.dense([0.0, 1.0, -2.0, 3.0]),),    (Vectors.dense([-1.0, 2.0, 4.0, -7.0]),),    (Vectors.dense([14.0, -2.0, -5.0, 1.0]),)], ["features"])# 实现了DCT- ii，并将结果缩放1/√2dct = DCT(inverse=False,           inputCol="features",           outputCol="featuresDCT")dctDf = dct.transform(df)dctDf.select("featuresDCT").show(truncate=False)

输出： 

+----------------------------------------------------------------+|featuresDCT                                                     |+----------------------------------------------------------------+|[1.0,-1.1480502970952693,2.0000000000000004,-2.7716385975338604]||[-1.0,3.378492794482933,-7.000000000000001,2.9301512653149677]  ||[4.0,9.304453421915744,11.000000000000002,1.5579302036357163]   |+----------------------------------------------------------------+

StringIndexer

StringIndexer 将标签的字符串列编码为标签索引的列。 StringIndexer 可以编码多列。

索引在 [0, numLabels) 中支持四种排序选项：

  • “frequencyDesc”：按标签频率降序排列（最频繁的标签分配为 0）
  • “frequencyAsc”：按标签频率升序排列（最不频繁的标签分配为 0）
  • “alphabetDesc”：字母降序
  • “alphabetAsc”：字母升序（默认 = “frequencyDesc”） 
from pyspark.ml.feature import StringIndexerdf = spark.createDataFrame(    [(0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c")],    ["id", "category"])indexer = StringIndexer(inputCol="category",                         outputCol="categoryIndex")indexed = indexer.fit(df).transform(df)indexed.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|+---+--------+-------------+

IndexToString

对称地 StringIndexer ， IndexToString 将一列标签索引映射回包含作为字符串的原始标签的列。 

from pyspark.ml.feature import IndexToString, StringIndexerdf = spark.createDataFrame(    [(0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c")],    ["id", "category"])indexer = StringIndexer(inputCol="category",                         outputCol="categoryIndex")model = indexer.fit(df)indexed = model.transform(df)print("转换字符串列 '%s' 到索引列 '%s'"      % (indexer.getInputCol(), indexer.getOutputCol()))indexed.show()print("StringIndexer将在输出列元数据中存储标签\n")converter = IndexToString(inputCol="categoryIndex", outputCol="originalCategory")converted = converter.transform(indexed)print("在元数据中使用标签，将转换索引列 '%s' 返回到原始的字符串列 '%s'"      % (converter.getInputCol(), converter.getOutputCol()))converted.select("id", "categoryIndex", "originalCategory").show()

输出： 

转换字符串列 'category' 到索引列 'categoryIndex'+---+--------+-------------+| 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|+---+--------+-------------+StringIndexer将在输出列元数据中存储标签在元数据中使用标签，将转换索引列 'categoryIndex' 返回到原始的字符串列 'originalCategory'+---+-------------+----------------+| id|categoryIndex|originalCategory|+---+-------------+----------------+|  0|          0.0|               a||  1|          2.0|               b||  2|          1.0|               c||  3|          0.0|               a||  4|          0.0|               a||  5|          1.0|               c|+---+-------------+----------------+

OneHotEncoder

One-hot 编码将表示为标签索引的分类特征映射到二进制向量，该向量最多具有一个单值，表示所有特征值集中存在特定特征值。这种编码允许期望连续特征的算法（例如逻辑回归）使用分类特征。对于字符串类型的输入数据，通常首先使用StringIndexer对分类特征进行编码。

OneHotEncoder 可以转换多个列，为每个输入列返回一个单热编码的输出向量列。通常使用VectorAssembler将这些向量合并为单个特征向量。 

from pyspark.ml.feature import OneHotEncoderdf = spark.createDataFrame([    (0.0, 1.0),    (1.0, 0.0),    (2.0, 1.0),    (0.0, 2.0),    (0.0, 1.0),    (2.0, 0.0)], ["categoryIndex1", "categoryIndex2"])encoder = OneHotEncoder(inputCols=["categoryIndex1", "categoryIndex2"],                        outputCols=["categoryVec1", "categoryVec2"])# `OneHotEncoder`支持`handleInvalid`参数选择在转换数据过程中如何处理无效输入# 可用选项包括"keep"（任何无效输入都分配给额外的分类索引）和"error"（抛出错误）。model = encoder.fit(df)encoded = model.transform(df)encoded.show()

输出： 

+--------------+--------------+-------------+-------------+|categoryIndex1|categoryIndex2| categoryVec1| categoryVec2|+--------------+--------------+-------------+-------------+|           0.0|           1.0|(2,[0],[1.0])|(2,[1],[1.0])||           1.0|           0.0|(2,[1],[1.0])|(2,[0],[1.0])||           2.0|           1.0|    (2,[],[])|(2,[1],[1.0])||           0.0|           2.0|(2,[0],[1.0])|    (2,[],[])||           0.0|           1.0|(2,[0],[1.0])|(2,[1],[1.0])||           2.0|           0.0|    (2,[],[])|(2,[0],[1.0])|+--------------+--------------+-------------+-------------+

VectorIndexer

VectorIndexer 向量索引器帮助索引向量数据集中的分类特征。它可以自动决定哪些特征是分类的，并将原始值转换为分类索引。具体来说，它做了以下操作:

  1. 接受一个Vector类型的输入列和一个参数 maxCategories 。
  2. 根据不同值的数量决定哪些特征应该是分类的，其中最多 maxCategories 声明的特征是分类的。
  3. 为每个分类特征计算基于 0 的分类索引。
  4. 索引分类特征并将原始特征值转换为索引。

索引分类特征允许决策树和树集成等算法适当地处理分类特征，从而提高性能。 

from pyspark.ml.feature import VectorIndexerdata = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")indexer = VectorIndexer(inputCol="features",                         outputCol="indexed",                         maxCategories=10)indexerModel = indexer.fit(data)categoricalFeatures = indexerModel.categoryMapsprint("Chose %d categorical features: %s" %      (len(categoricalFeatures), ", ".join(str(k) for k in categoricalFeatures.keys())))# 创建新列“索引”，将分类值转换为索引indexedData = indexerModel.transform(data)indexedData.show()

输出： 

Chose 351 categorical features: 645, 69, 365, 138, 101, 479, 333, 249, ...+-----+--------------------+--------------------+|label|            features|             indexed|+-----+--------------------+--------------------+|  0.0|(692,[127,128,129...|(692,[127,128,129...||  1.0|(692,[158,159,160...|(692,[158,159,160...||  1.0|(692,[124,125,126...|(692,[124,125,126...||  1.0|(692,[152,153,154...|(692,[152,153,154...||  1.0|(692,[151,152,153...|(692,[151,152,153...|...|  1.0|(692,[150,151,152...|(692,[150,151,152...||  0.0|(692,[124,125,126...|(692,[124,125,126...||  0.0|(692,[152,153,154...|(692,[152,153,154...||  1.0|(692,[97,98,99,12...|(692,[97,98,99,12...||  1.0|(692,[124,125,126...|(692,[124,125,126...|+-----+--------------------+--------------------+only showing top 20 rows

Interaction

Interaction 是一个 Transformer 采用向量或双值列，并生成一个向量列，其中包含来自每个输入列的一个值的所有组合的乘积。 

from pyspark.ml.feature import Interaction, VectorAssemblerdf = spark.createDataFrame(    [(1, 1, 2, 3, 8, 4, 5),     (2, 4, 3, 8, 7, 9, 8),     (3, 6, 1, 9, 2, 3, 6),     (4, 10, 8, 6, 9, 4, 5),     (5, 9, 2, 7, 10, 7, 3),     (6, 1, 1, 4, 2, 8, 4)],    ["id1", "id2", "id3", "id4", "id5", "id6", "id7"])assembler1 = VectorAssembler(inputCols=["id2", "id3", "id4"], outputCol="vec1")assembled1 = assembler1.transform(df)assembler2 = VectorAssembler(inputCols=["id5", "id6", "id7"], outputCol="vec2")assembled2 = assembler2.transform(assembled1).select("id1", "vec1", "vec2")interaction = Interaction(inputCols=["id1", "vec1", "vec2"], outputCol="interactedCol")interacted = interaction.transform(assembled2)interacted.show(truncate=False)

输出： 

+---+--------------+--------------+------------------------------------------------------+|id1|vec1          |vec2          |interactedCol                                         |+---+--------------+--------------+------------------------------------------------------+|1  |[1.0,2.0,3.0] |[8.0,4.0,5.0] |[8.0,4.0,5.0,16.0,8.0,10.0,24.0,12.0,15.0]            ||2  |[4.0,3.0,8.0] |[7.0,9.0,8.0] |[56.0,72.0,64.0,42.0,54.0,48.0,112.0,144.0,128.0]     ||3  |[6.0,1.0,9.0] |[2.0,3.0,6.0] |[36.0,54.0,108.0,6.0,9.0,18.0,54.0,81.0,162.0]        ||4  |[10.0,8.0,6.0]|[9.0,4.0,5.0] |[360.0,160.0,200.0,288.0,128.0,160.0,216.0,96.0,120.0]||5  |[9.0,2.0,7.0] |[10.0,7.0,3.0]|[450.0,315.0,135.0,100.0,70.0,30.0,350.0,245.0,105.0] ||6  |[1.0,1.0,4.0] |[2.0,8.0,4.0] |[12.0,48.0,24.0,12.0,48.0,24.0,48.0,192.0,96.0]       |+---+--------------+--------------+------------------------------------------------------+
🏴‍☠️宝藏级🏴‍☠️ 原创公众号『 数据STUDIO 』内容超级硬核。公众号以Python为核心语言，垂直于数据科学领域，包括 可戳 👉 Python ｜ MySQL ｜ 数据分析 ｜ 数据可视化 ｜ 机器学习与数据挖掘 ｜ 爬虫 等，从入门到进阶！

长按👇关注- 数据STUDIO -设为星标，干货速递

KEEP THE THREAD GOING

更多来源文章