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

Question

我有一个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包含上述函数输出的新列(或类似功能的东西).

Answer 1

Spark> = 2.3

因为Spark 2.3 OneHotEncoder被弃用了OneHotEncoderEstimator.如果您使用最新版本,请修改encoder代码

from pyspark.ml.feature import OneHotEncoderEstimator

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

Spark <2.3

好吧,你可以写一个UDF,但为什么呢？已经有很多工具可以用来处理这类任务:

from pyspark.ml.feature import OneHotEncoder

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

首先OneHotEncoder.

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

下一个StringIndexer:

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包含分类变量,您可以使用它VectorAssembler来设置所需的元数据:

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])|
## +------+----+--------------+-------------+

但事实并非如此.

最后,您可以使用管道包装所有这些:

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.

获得相当的输出的另一种方式是StringIndexer 其中:

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

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)

关于你的问题:

制作一个具有类似功能的UDF,我可以在Spark SQL查询中使用(或者我想其他方式)

它就像任何其他UDF一样.确保你使用支持的类型,以及一切都应该工作得很好.

获取上述地图产生的RDD,并将其作为新列添加到user_data数据框中？

from pyspark.ml.feature import RFormula

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

注意:

对于Spark 1.x替换RFormula为StringIndexer.