将 StringIndexer 应用于 PySpark Dataframe 中的多个列答案

【问题标题】：Apply StringIndexer to several columns in a PySpark Dataframe将 StringIndexer 应用于 PySpark Dataframe 中的多个列
【发布时间】：2016-08-24 20:16:36
【问题描述】：

我有一个 PySpark 数据框

+-------+--------------+----+----+
|address|          date|name|food|
+-------+--------------+----+----+
|1111111|20151122045510| Yin|gre |
|1111111|20151122045501| Yin|gre |
|1111111|20151122045500| Yln|gra |
|1111112|20151122065832| Yun|ddd |
|1111113|20160101003221| Yan|fdf |
|1111111|20160703045231| Yin|gre |
|1111114|20150419134543| Yin|fdf |
|1111115|20151123174302| Yen|ddd |
|2111115|      20123192| Yen|gre |
+-------+--------------+----+----+

我想转换为与 pyspark.ml 一起使用。我可以使用 StringIndexer 将名称列转换为数字类别：

indexer = StringIndexer(inputCol="name", outputCol="name_index").fit(df)
df_ind = indexer.transform(df)
df_ind.show()
+-------+--------------+----+----------+----+
|address|          date|name|name_index|food|
+-------+--------------+----+----------+----+
|1111111|20151122045510| Yin|       0.0|gre |
|1111111|20151122045501| Yin|       0.0|gre |
|1111111|20151122045500| Yln|       2.0|gra |
|1111112|20151122065832| Yun|       4.0|ddd |
|1111113|20160101003221| Yan|       3.0|fdf |
|1111111|20160703045231| Yin|       0.0|gre |
|1111114|20150419134543| Yin|       0.0|fdf |
|1111115|20151123174302| Yen|       1.0|ddd |
|2111115|      20123192| Yen|       1.0|gre |
+-------+--------------+----+----------+----+

如何使用 StringIndexer 转换多个列（例如，name 和 food，每个都有自己的 StringIndexer），然后使用 VectorAssembler 生成特征向量？还是我必须为每列创建一个StringIndexer？

** 编辑**：这不是骗人的，因为我需要以编程方式对具有不同列名的多个数据框进行此操作。我不能使用VectorIndexer 或VectorAssembler，因为这些列不是数字。

** EDIT 2**: 一个暂定的解决方案是

indexers = [StringIndexer(inputCol=column, outputCol=column+"_index").fit(df).transform(df) for column in df.columns ]

我现在创建一个包含三个数据框的列表，每个数据框都与原始数据框和转换后的列相同。现在我需要加入然后形成最终的数据帧，但这非常低效。

【问题讨论】：

Encode and assemble multiple features in PySpark的可能重复
类似但不是真的。他为每个字符串索引器一次做一列，我需要同时对几列做，而不是每一个分开
那就不可能了。甚至输出是什么？
好的，这就是我想要的。然后我将编写一个 UDF。
你不想要更多类似CountVectroizer的东西吗？

标签： python apache-spark pyspark

【解决方案1】：

我发现的最佳方法是将多个StringIndex 组合到一个列表中，然后使用Pipeline 来执行它们：

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer

indexers = [StringIndexer(inputCol=column, outputCol=column+"_index").fit(df) for column in list(set(df.columns)-set(['date'])) ]


pipeline = Pipeline(stages=indexers)
df_r = pipeline.fit(df).transform(df)

df_r.show()
+-------+--------------+----+----+----------+----------+-------------+
|address|          date|food|name|food_index|name_index|address_index|
+-------+--------------+----+----+----------+----------+-------------+
|1111111|20151122045510| gre| Yin|       0.0|       0.0|          0.0|
|1111111|20151122045501| gra| Yin|       2.0|       0.0|          0.0|
|1111111|20151122045500| gre| Yln|       0.0|       2.0|          0.0|
|1111112|20151122065832| gre| Yun|       0.0|       4.0|          3.0|
|1111113|20160101003221| gre| Yan|       0.0|       3.0|          1.0|
|1111111|20160703045231| gre| Yin|       0.0|       0.0|          0.0|
|1111114|20150419134543| gre| Yin|       0.0|       0.0|          5.0|
|1111115|20151123174302| ddd| Yen|       1.0|       1.0|          2.0|
|2111115|      20123192| ddd| Yen|       1.0|       1.0|          4.0|
+-------+--------------+----+----+----------+----------+-------------+

【讨论】：

我有一个迭代循环，里面有重新分配。宇宙中最丑陋的代码.. 猜猜是时候投入一点精力并一劳永逸地切换到Pipelines了！
你真的需要fit in indexers 吗？无论如何，你在pipeline 中运行fit。
我试过这个解决方案，但我发现这与for循环没有区别。我仍然有多个 spark-job 作为 for-loop

【解决方案2】：

我可以为您提供以下解决方案。最好使用管道在更大的数据集上进行此类转换。它们还使您的代码更易于遵循和理解。如果需要，您可以向管道添加更多阶段。例如添加一个编码器。

#create a list of the columns that are string typed
categoricalColumns = [item[0] for item in df.dtypes if item[1].startswith('string') ]

#define a list of stages in your pipeline. The string indexer will be one stage
stages = []

#iterate through all categorical values
for categoricalCol in categoricalColumns:
    #create a string indexer for those categorical values and assign a new name including the word 'Index'
    stringIndexer = StringIndexer(inputCol = categoricalCol, outputCol = categoricalCol + 'Index')

    #append the string Indexer to our list of stages
    stages += [stringIndexer]

#Create the pipeline. Assign the satges list to the pipeline key word stages
pipeline = Pipeline(stages = stages)
#fit the pipeline to our dataframe
pipelineModel = pipeline.fit(df)
#transform the dataframe
df= pipelineModel.transform(df)

请看看我的reference

【讨论】：

【解决方案3】：

使用 PySpark 3.0+，现在这更容易了，您可以使用 inputCols 和 outputCols 选项： https://spark.apache.org/docs/latest/ml-features#stringindexer

class pyspark.ml.feature.StringIndexer(inputCol=None, outputCol=None, inputCols=None, outputCols=None, handleInvalid='error', stringOrderType='frequencyDesc')

【讨论】：

您的链接已损坏。此外，很高兴看到一个关于如何在此处使用这些新选项的示例。
链接已修复 :)

【解决方案4】：

将 StringIndexer 应用于 PySpark Dataframe 中的多个列对于火花 2.4.7

from pyspark.ml.feature import StringIndexer
from pyspark.ml import Pipeline

indexers = [StringIndexer(inputCol="F1", outputCol="F1Index") , StringIndexer(inputCol="F5", outputCol="F5Index")]


pipeline = Pipeline(stages=indexers)
DF6 = pipeline.fit(DF5).transform(DF5)

DF6.show()

【讨论】：