在Spark中如何使用UDO作为参数调用UDF以避免二进制错误

Posted 2021-04-19

我用UDO作为参数定义了一个UDF。但是当我试图在数据框中调用它时，我收到错误消息“org.apache.spark.SparkException：无法执行用户定义的函数（$ anonfun $ 1：（array）=> int）”。只是想知道预期异常提到UDO为二进制，还应该如何修复它？

val logCount = (logs: util.List[LogRecord]) => logs.size()
val logCountUdf = udf(logCount)
// The column 'LogRecords' is the agg function collect_list of UDO LogRecord
df.withColumn("LogCount", logCountUdf($"LogRecords"))

答案

通常，您无法将自定义对象传递给UDF，并且您应该只为非空行调用udf，否则UDF中会出现NullPointerException。尝试：

val logCount = (logs: Seq[Row]) => logs.size()
val logCountUdf = udf(logCount)

df.withColumn("LogCount", when($"LogRecords".isNotNull,logCountUdf($"LogRecords")))

或者只是使用内置函数size来获取logCount：

df.withColumn("LogCount", size($"LogRecords"))