在 PySpark 中编码和组装多个功能 [英] Encode and assemble multiple features in PySpark
问题描述
我有一个 Python 类,用于在 Spark 中加载和处理一些数据.在我需要做的各种事情中,我正在生成一个从 Spark 数据帧中的各个列派生的虚拟变量列表.我的问题是我不确定如何正确定义用户定义的函数来完成我需要的功能.
I have a Python class that I'm using to load and process some data in Spark. Among various things I need to do, I'm generating a list of dummy variables derived from various columns in a Spark dataframe. My problem is that I'm not sure how to properly define a User Defined Function to accomplish what I need.
我确实目前有一个方法,当映射到底层数据帧 RDD 时,解决了一半的问题(请记住,这是一个更大的 data_processor
类中的方法):
I do currently have a method that, when mapped over the underlying dataframe RDD, solves half the problem (remember that this is a method in a larger data_processor
class):
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
本质上,对于指定的数据框,它的作用是获取指定列的分类变量值,并返回这些新虚拟变量的值列表.这意味着以下代码:
Essentially what this does is, for the specified dataframe, takes the categorical variable values for the specified columns, and returns a list of the values of these new dummy variables. That means the following code:
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) 制作一个可以在 Spark SQL 查询中使用的具有类似功能的 UDF(或其他方式,我想),或者(b)采用上述地图产生的 RDD 并将其作为新列添加到 user_data 数据帧?
This is exactly what I want in terms of generating the list of dummy variables I want, but here's my question: How can I either (a) make a UDF with similar functionality that I can use in a Spark SQL query (or some other way, I suppose), or (b) take the RDD resulting from the map described above and add it as a new column to the user_data dataframe?
无论哪种方式,我需要做的是生成一个包含 user_data 列的新数据框,以及一个包含上述函数输出的新列(我们称之为 feature_array
)(或其他功能等效).
Either way, what I need to do is generate a new dataframe containing the columns from user_data, along with a new column (let's call it feature_array
) containing the output of the function above (or something functionally equivalent).
推荐答案
Spark >= 2.3, >= 3.0
自从 Spark 2.3 OneHotEncoder
被弃用,取而代之的是 OneHotEncoderEstimator
.如果你使用最近的版本请修改encoder
代码
Since Spark 2.3 OneHotEncoder
is deprecated in favor of OneHotEncoderEstimator
. If you use a recent release please modify encoder
code
from pyspark.ml.feature import OneHotEncoderEstimator
encoder = OneHotEncoderEstimator(
inputCols=["gender_numeric"],
outputCols=["gender_vector"]
)
在 Spark 3.0 中,此变体已重命名为 OneHotEncoder
:
In Spark 3.0 this variant has been renamed to OneHotEncoder
:
from pyspark.ml.feature import OneHotEncoder
encoder = OneHotEncoder(
inputCols=["gender_numeric"],
outputCols=["gender_vector"]
)
另外 StringIndexer
已扩展为支持多个输入列:
Additionally StringIndexer
has been extended to support multiple input columns:
StringIndexer(inputCols=["gender"], outputCols=["gender_numeric"])
火花<2.3
好吧,您可以编写 UDF,但为什么要这样做?已经有很多工具可以处理此类任务:
Well, you can write an UDF but why would you? There are already quite a few tools designed to handle this category of tasks:
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
来设置所需的元数据:
If bar
contained categorical variables you could use VectorIndexer
to set required metadata:
from pyspark.ml.feature import VectorIndexer
vector_indexer = VectorIndexer(inputCol="bar", outputCol="bar_indexed")
但这里不是这样.
最后,您可以使用管道包装所有这些:
Finally you can wrap all of that using pipelines:
from pyspark.ml import Pipeline
pipeline = Pipeline(stages=[indexer, encoder, vector_indexer, assembler])
model = pipeline.fit(df)
transformed = model.transform(df)
可以说,它比从头开始编写所有内容都要健壮和干净.有一些警告,尤其是当您需要在不同数据集之间进行一致编码时.您可以在 StringIndexer
和 VectorIndexer
的官方文档中阅读更多内容.
Arguably it is much robust and clean approach than writing everything from scratch. There are some caveats especially when you need consistent encoding between different datasets. You can read more in the official documentation for StringIndexer
and VectorIndexer
.
获得类似输出的另一种方法是 RFormula
其中:
Another way to get a comparable output is RFormula
which:
RFormula
生成特征的向量列和标签的双列或字符串列.就像在 R 中使用公式进行线性回归一样,字符串输入列将被单热编码,数字列将被转换为双精度.如果标签列是字符串类型,它会首先用 StringIndexer
转换为 double.如果 DataFrame 中不存在标签列,则将根据公式中指定的响应变量创建输出标签列.
RFormula
produces a vector column of features and a double or string column of label. Like when formulas are used in R for linear regression, string input columns will be one-hot encoded, and numeric columns will be cast to doubles. If the label column is of type string, it will be first transformed to double withStringIndexer
. If the label column does not exist in the DataFrame, the output label column will be created from the specified response variable in the formula.
from pyspark.ml.feature import RFormula
rf = RFormula(formula="~ gender + bar + foo - 1")
final_df_rf = rf.fit(df).transform(df)
如您所见,它更简洁,但更难编写,不允许进行太多自定义.然而,像这样的简单管道的结果将是相同的:
As you can see it is much more concise, but harder to compose doesn't allow much customization. Nevertheless the result for a simple pipeline like this one will be identical:
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 查询中使用它(或其他方式,我想)
make a UDF with similar functionality that I can use in a Spark SQL query (or some other way, I suppose)
它只是一个像其他任何一样的 UDF.确保您使用支持的类型,除此之外,一切都应该正常工作.
It is just an UDF like any other. Make sure you use supported types and beyond that everything should work just fine.
将上述地图生成的 RDD 作为新列添加到 user_data 数据框中?
take the RDD resulting from the map described above and add it as a new column to the user_data dataframe?
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
.
For Spark 1.x replace pyspark.ml.linalg
with pyspark.mllib.linalg
.
这篇关于在 PySpark 中编码和组装多个功能的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!