在PySpark中对多个特征进行编码和组装

Question

在PySpark中对多个特征进行编码和组装

pythonapache-sparkapache-spark-sqlapache-spark-mllibapache-spark-ml

27

我有一个Python类，用于在Spark中加载和处理一些数据。在需要完成的各种任务中，我正在生成从Spark dataframe中的各列派生的虚拟变量列表。我的问题是，我不确定如何正确定义用户定义的函数来实现我所需的功能。

我目前有一个方法，可以在底层dataframe RDD上映射时解决一半的问题（请记住，这是大型data_processor类中的一个方法）：

def build_feature_arr(self,table):
    # this dict has keys for all the columns for which I need dummy coding
    categories = {'gender':['1','2'], ..}

    # there are actually two differnt dataframes that I need to do this for, this just specifies which I'm looking at, and grabs the relevant features from a config file
    if table == 'users':
        iter_over = self.config.dyadic_features_to_include
    elif table == 'activty':
        iter_over = self.config.user_features_to_include

    def _build_feature_arr(row):
        result = []
        row = row.asDict()
        for col in iter_over:
            column_value = str(row[col]).lower()
            cats = categories[col]
            result += [1 if column_value and cat==column_value else 0 for cat in cats]
        return result
    return _build_feature_arr

本质上，这个函数针对指定的数据框架，获取指定列的分类变量值，并返回这些新虚拟变量的值列表。这意味着以下代码：

data = data_processor(init_args)
result = data.user_data.rdd.map(self.build_feature_arr('users'))

返回类似于以下内容:

In [39]: result.take(10)
Out[39]:
[[1, 0, 0, 0, 1, 0],
 [1, 0, 0, 1, 0, 0],
 [1, 0, 0, 0, 0, 0],
 [1, 0, 1, 0, 0, 0],
 [1, 0, 0, 1, 0, 0],
 [1, 0, 0, 1, 0, 0],
 [0, 1, 1, 0, 0, 0],
 [1, 0, 1, 1, 0, 0],
 [1, 0, 0, 1, 0, 0],
 [1, 0, 0, 0, 0, 1]]

这正是我想要的生成虚拟变量列表的方法，但是我的问题在于：我该如何（a）创建一个具有类似功能的UDF，以便在Spark SQL查询中使用它（或者其他方式），还是（b）将上述map操作的RDD作为新列添加到user_data数据帧中呢？

无论哪种方式，我需要生成一个新数据帧，其中包含来自user_data的列，以及一个新列（称为feature_array），其中包含上述函数的输出（或者与之等效的内容）。

- moustachio

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- zero323 · Accepted Answer

Spark >= 2.3, >= 3.0

自 Spark 2.3 起，OneHotEncoder 已被弃用，建议使用 OneHotEncoderEstimator。如果您使用的是最新版本，请修改 encoder 代码。

from pyspark.ml.feature import OneHotEncoderEstimator

encoder = OneHotEncoderEstimator(
    inputCols=["gender_numeric"],  
    outputCols=["gender_vector"]
)

在Spark 3.0中，这个变量已被重命名为OneHotEncoder:

from pyspark.ml.feature import OneHotEncoder

encoder = OneHotEncoder(
    inputCols=["gender_numeric"],  
    outputCols=["gender_vector"]
)

此外，StringIndexer 已扩展以支持多个输入列：

StringIndexer(inputCols=["gender"], outputCols=["gender_numeric"])

Spark < 2.3

你可以编写一个UDF，但为什么要这样做呢？已经有很多工具专门处理这类任务：

from pyspark.sql import Row
from pyspark.ml.linalg import DenseVector

row = Row("gender", "foo", "bar")

df = sc.parallelize([
  row("0", 3.0, DenseVector([0, 2.1, 1.0])),
  row("1", 1.0, DenseVector([0, 1.1, 1.0])),
  row("1", -1.0, DenseVector([0, 3.4, 0.0])),
  row("0", -3.0, DenseVector([0, 4.1, 0.0]))
]).toDF()

首先是StringIndexer。

from pyspark.ml.feature import StringIndexer

indexer = StringIndexer(inputCol="gender", outputCol="gender_numeric").fit(df)
indexed_df = indexer.transform(df)
indexed_df.drop("bar").show()

## +------+----+--------------+
## |gender| foo|gender_numeric|
## +------+----+--------------+
## |     0| 3.0|           0.0|
## |     1| 1.0|           1.0|
## |     1|-1.0|           1.0|
## |     0|-3.0|           0.0|
## +------+----+--------------+

下一个OneHotEncoder:

from pyspark.ml.feature import OneHotEncoder

encoder = OneHotEncoder(inputCol="gender_numeric", outputCol="gender_vector")
encoded_df = encoder.transform(indexed_df)
encoded_df.drop("bar").show()

## +------+----+--------------+-------------+
## |gender| foo|gender_numeric|gender_vector|
## +------+----+--------------+-------------+
## |     0| 3.0|           0.0|(1,[0],[1.0])|
## |     1| 1.0|           1.0|    (1,[],[])|
## |     1|-1.0|           1.0|    (1,[],[])|
## |     0|-3.0|           0.0|(1,[0],[1.0])|
## +------+----+--------------+-------------+

VectorAssembler：

from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(
    inputCols=["gender_vector", "bar", "foo"], outputCol="features")

encoded_df_with_indexed_bar = (vector_indexer
    .fit(encoded_df)
    .transform(encoded_df))

final_df = assembler.transform(encoded_df)

如果bar包含分类变量，您可以使用VectorIndexer来设置所需的元数据：

from pyspark.ml.feature import VectorIndexer

vector_indexer = VectorIndexer(inputCol="bar", outputCol="bar_indexed")

但这里并非如此。

最后，您可以使用管道将所有内容包装起来：

from pyspark.ml import Pipeline
pipeline = Pipeline(stages=[indexer, encoder, vector_indexer, assembler])
model = pipeline.fit(df)
transformed = model.transform(df)

Arguably，这种方法比从头开始编写更健壮、更清晰。但是需要注意的是，当您需要在不同数据集之间保持一致的编码时，有一些注意事项。您可以在官方文档中了解有关StringIndexer和VectorIndexer的更多信息。

另一种获得可比较输出的方法是使用RFormula：

RFormula生成一个特征向量列和一个双精度或字符串列的标签。就像在R中用于线性回归的公式一样，字符串输入列将进行单热编码，数字列将转换为双精度。如果标签列的类型为字符串，则会首先使用StringIndexer将其转换为双精度。如果DataFrame中不存在标签列，则将从公式中指定的响应变量创建输出标签列。

from pyspark.ml.feature import RFormula

rf = RFormula(formula="~ gender +  bar + foo - 1")
final_df_rf = rf.fit(df).transform(df)

正如您所看到的，这种方法更加简洁，但更难组合，也不允许太多自定义。然而，对于像这样简单的管道来说，结果将是相同的：

final_df_rf.select("features").show(4, False)

## +----------------------+
## |features              |
## +----------------------+
## |[1.0,0.0,2.1,1.0,3.0] |
## |[0.0,0.0,1.1,1.0,1.0] |
## |(5,[2,4],[3.4,-1.0])  |
## |[1.0,0.0,4.1,0.0,-3.0]|
## +----------------------+


final_df.select("features").show(4, False)

## +----------------------+
## |features              |
## +----------------------+
## |[1.0,0.0,2.1,1.0,3.0] |
## |[0.0,0.0,1.1,1.0,1.0] |
## |(5,[2,4],[3.4,-1.0])  |
## |[1.0,0.0,4.1,0.0,-3.0]|
## +----------------------+

关于您的问题：

创建一个具有类似功能的UDF，我可以在Spark SQL查询中使用它（或者其他方式）

这只是一个像其他UDF一样的函数。确保使用支持的类型，除此之外，一切都应该正常工作。

将从上述映射结果得到的RDD添加为user_data数据帧的新列？

from pyspark.ml.linalg import VectorUDT
from pyspark.sql.types import StructType, StructField

schema = StructType([StructField("features", VectorUDT(), True)])
row = Row("features")
result.map(lambda x: row(DenseVector(x))).toDF(schema)

注意：

对于 Spark 1.x，请将 pyspark.ml.linalg 替换为 pyspark.mllib.linalg。