python - PySpark按行功能组合

举一个简化的例子，我有一个数据框“df”，其列为“col1，col2”，我想在对每列应用一个函数后计算一个按行的最大值:

def f(x):
    return (x+1)

max_udf=udf(lambda x,y: max(x,y), IntegerType())
f_udf=udf(f, IntegerType())

df2=df.withColumn("result", max_udf(f_udf(df.col1),f_udf(df.col2)))

因此，如果df:

col1   col2
1      2
3      0

然后

df2:

col1   col2  result
1      2     3
3      0     4

上面的方法似乎无效，并产生“无法评估表达式:PythonUDF#f ...”

我绝对肯定“f_udf”在我的桌子上可以正常工作，而主要问题在于max_udf。

如果不创建额外的列或使用基本的map/reduce，是否可以完全使用数据框和udfs来完成上述操作？我应该如何修改“max_udf”？

我也尝试过:

max_udf=udf(max, IntegerType())

产生相同的错误。

我还确认了以下作品:

df2=(df.withColumn("temp1", f_udf(df.col1))
       .withColumn("temp2", f_udf(df.col2))

df2=df2.withColumn("result", max_udf(df2.temp1,df2.temp2))

为什么我不能一口气做到这些？

我想看到一个概括为任何函数“f_udf”和“max_udf”的答案。

最佳答案

我遇到了类似的问题，并在this stackoverflow question的答案中找到了解决方案

要将多列或整行传递给UDF，请使用struct:

from pyspark.sql.functions import udf, struct
from pyspark.sql.types import IntegerType

df = sqlContext.createDataFrame([(None, None), (1, None), (None, 2)], ("a", "b"))

count_empty_columns = udf(lambda row: len([x for x in row if x == None]), IntegerType())

new_df = df.withColumn("null_count", count_empty_columns(struct([df[x] for x in df.columns])))

new_df.show()

返回:

+----+----+----------+
|   a|   b|null_count|
+----+----+----------+
|null|null|         2|
|   1|null|         1|
|null|   2|         1|
+----+----+----------+

关于python - PySpark按行功能组合，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/36584812/