如何将函数应用于 PySpark DataFrame 指定列的每一行答案

【问题标题】：How to apply function to each row of specified column of PySpark DataFrame如何将函数应用于 PySpark DataFrame 指定列的每一行
【发布时间】：2017-07-17 01:44:07
【问题描述】：

我有一个 PySpark DataFrame 由三列组成，其结构如下。

In[1]: df.take(1)    
Out[1]:
[Row(angle_est=-0.006815859163590619, rwsep_est=0.00019571401752467945, cost_est=34.33651951754235)]

我想要做的是检索第一列 (angle_est) 的每个值，并将其作为参数 xMisallignment 传递给定义的函数以设置类对象的特定属性。定义的函数是：

def setMisAllignment(self, xMisallignment):
    if np.abs(xMisallignment) > 0.8:
       warnings.warn('You might set misallignment angle too large.')
    self.MisAllignment = xMisallignment

我正在尝试选择第一列并将其转换为 rdd，并将上述函数应用于 map() 函数，但它似乎不起作用，MisAllignment 无论如何都没有改变。

df.select(df.angle_est).rdd.map(lambda row: model0.setMisAllignment(row))

In[2]: model0.MisAllignment
Out[2]: 0.00111511718224

任何人有想法可以帮助我让该功能发挥作用吗？提前致谢！

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql spark-dataframe pyspark-sql

【解决方案1】：

您可以将您的函数注册为 spark UDF，类似于以下内容：

spark.udf.register("misallign", setMisAllignment)

您可以在此测试套件中获得许多创建和注册 UDF 的示例： https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/JavaUDFSuite.java

希望它能回答你的问题

【讨论】：