对 SparkMlib 中的几个分类列应用 OneHotEncoder [英] apply OneHotEncoder for several categorical columns in SparkMlib

问题描述

我有几个分类特征，并希望使用 OneHotEncoder 将它们全部转换.但是，当我尝试应用 StringIndexer 时，出现错误:

I have several categorical features and would like to transform them all using OneHotEncoder. However, when I tried to apply the StringIndexer, there I get an error:

stringIndexer = StringIndexer(
    inputCol = ['a', 'b','c','d'],
    outputCol = ['a_index', 'b_index','c_index','d_index']
)  

model = stringIndexer.fit(Data)

An error occurred while calling o328.fit.
: java.lang.ClassCastException: java.util.ArrayList cannot be cast to java.lang.String
    at org.apache.spark.ml.feature.StringIndexer.fit(StringIndexer.scala:79)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
    at py4j.Gateway.invoke(Gateway.java:259)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:207)
    at java.lang.Thread.run(Thread.java:745)

Traceback (most recent call last):
Py4JJavaError: An error occurred while calling o328.fit.
: java.lang.ClassCastException: java.util.ArrayList cannot be cast to java.lang.String
    at org.apache.spark.ml.feature.StringIndexer.fit(StringIndexer.scala:79)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
    at py4j.Gateway.invoke(Gateway.java:259)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:207)
    at java.lang.Thread.run(Thread.java:745)

推荐答案

Spark >= 3.0:

在 Spark 3.0 中 OneHotEncoderEstimator 已重命名为 OneHotEncoder:

In Spark 3.0 OneHotEncoderEstimator has been renamed to OneHotEncoder:

from pyspark.ml.feature import OneHotEncoderEstimator, OneHotEncoderModel

encoder = OneHotEncoderEstimator(...)

与

from pyspark.ml.feature import OneHotEncoder, OneHotEncoderModel

encoder = OneHotEncoder(...)

Spark >= 2.3

您可以使用新添加的OneHotEncoderEstimator:

from pyspark.ml.feature import OneHotEncoderEstimator, OneHotEncoderModel

encoder = OneHotEncoderEstimator(
    inputCols=[indexer.getOutputCol() for indexer in indexers],
    outputCols=[
        "{0}_encoded".format(indexer.getOutputCol()) for indexer in indexers]
)

assembler = VectorAssembler(
    inputCols=encoder.getOutputCols(),
    outputCol="features"
)

pipeline = Pipeline(stages=indexers + [encoder, assembler])
pipeline.fit(df).transform(df)

火花<2.3

这是不可能的.StringIndexer 转换器当时仅在单个列上运行，因此您需要为要转换的每一列使用单个索引器和单个编码器.

It is not possible. StringIndexer transformer operates only on a single column at the time so you'll need a single indexer and a single encoder for each column you want to transform.

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler

cols = ['a', 'b', 'c', 'd']

indexers = [
    StringIndexer(inputCol=c, outputCol="{0}_indexed".format(c))
    for c in cols
]

encoders = [
    OneHotEncoder(
        inputCol=indexer.getOutputCol(),
        outputCol="{0}_encoded".format(indexer.getOutputCol())) 
    for indexer in indexers
]

assembler = VectorAssembler(
    inputCols=[encoder.getOutputCol() for encoder in encoders],
    outputCol="features"
)


pipeline = Pipeline(stages=indexers + encoders + [assembler])
pipeline.fit(df).transform(df).show()

这篇关于对 SparkMlib 中的几个分类列应用 OneHotEncoder的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！