在 PySpark 中编码和组装多个功能

Posted 2023-03-12

【中文标题】在 PySpark 中编码和组装多个功能

【英文标题】：Encode and assemble multiple features in PySpark

【发布时间】：2016-01-04 02:32:55

【问题描述】：

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

我确实目前有一个方法，当映射到底层数据帧 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）从上述地图中获取 RDD 并将其作为新列添加到 user_data 数据帧？

无论哪种方式，我需要做的是生成一个新的数据框，其中包含来自 user_data 的列，以及一个包含上述函数输出（或功能等效）的新列（我们称之为feature_array）。

【参考方案1】：

火花 >= 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"])

火花

嗯，你可以写一个 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)

可以说，与从头开始编写所有内容相比，它更加健壮和简洁。有一些警告，尤其是当您需要在不同数据集之间进行一致编码时。您可以在StringIndexer 和VectorIndexer 的官方文档中阅读更多内容。

获得可比输出的另一种方法是RFormulawhich：

RFormula 生成特征向量列和标签的双精度列或字符串列。就像在 R 中使用公式进行线性回归一样，字符串输入列将被一次性编码，而数字列将被转换为双精度值。如果标签列是字符串类型，它将首先转换为带有StringIndexer 的double。如果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。

【讨论】：

@DavidArenburg 在这种特殊情况下，这是因为 OP 想要获得虚拟变量（如 R 中的 model.matrix）。最有可能训练某种类型的线性模型。 Rish 解释 - 字符串索引器从字符串创建类似因子的列，一个热调用 model.matrix :)

感谢@zero323！请注意：从 Spark 2.0+ 开始，from pyspark.mllib.linalg import DenseVector 应替换为 from pyspark.ml.linalg import DenseVector，否则您可能会在 VectorIndexer 阶段遇到类型错误

我有一个问题......如果我在这个数据上运行一个 randomforest_Classifier 我会得到随机森林叶子的数字（因为索引）。如何以简洁的方式将其与原始描述（即英文文本）联系起来。例如，随机森林分类器没有肉类数据，这成为一项艰巨的任务。我有一个模糊的想法，我必须使用 IndexToString() 之类的东西，但我不知道如何使用它

@zero323 你能解释一下如何在你的例子中读取oneHotEncoding之后的gender_vector吗？

@MichelleOwen 阅读是什么意思？如果是关于映射/解释，那么它就是元数据这个***.com/a/35305314/1560062 和这个github.com/awesome-spark/spark-gotchas/blob/… 应该会有所启发。