我有一个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)获取上述映射产生的RDD,并将其作为新列添加到user_data数据框架中?
无论采用哪种方法,我需要做的都是生成一个包含来自user_data的列的新数据帧,以及一个包含上述函数(或其他等效函数)输出的新列(让我们称其为feature_array
)。
发布于 2015-10-07 13:59:07
Spark >= 2.3,>= 3.0
从Spark2.3开始,为了支持OneHotEncoderEstimator
,OneHotEncoder
被弃用了。如果您使用最新版本,请修改encoder
代码
from pyspark.ml.feature import OneHotEncoderEstimator
encoder = OneHotEncoderEstimator(
inputCols=["gender_numeric"],
outputCols=["gender_vector"]
)
在Spark3.0中,这个变体已经被重命名为OneHotEncoder
from pyspark.ml.feature import OneHotEncoder
encoder = OneHotEncoder(
inputCols=["gender_numeric"],
outputCols=["gender_vector"]
)
此外,StringIndexer
已经扩展为支持多个输入列:
StringIndexer(inputCols=["gender"], outputCols=["gender_numeric"])
火花< 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)
可以说,与从头开始编写一切相比,它是一种更健壮和干净的方法。有一些注意事项,特别是当您需要不同数据集之间的一致编码时。您可以在StringIndexer
和VectorIndexer
的官方文档中阅读更多信息。
获得可比较输出的另一种方法是RFormula
which
RFormula
生成要素的矢量列和标签的双精度列或字符串列。就像在R中使用公式进行线性回归一样,字符串输入列将采用单热编码,而数值列将转换为双精度。如果标签列的类型为string,则首先使用StringIndexer
将其转换为double。如果DataFrame中不存在label列,将从公式中指定的response变量创建输出label列。
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
。
https://stackoverflow.com/questions/32982425
复制相似问题