星火UDF - 任务不能序列异常

Question

我想用下面的Scala代码来创建UDF

lazy val formattedDF = df.withColumn("result_col", validateudf(df("id")))

val validateudf = udf((id: Int) => {

  if(id == 1){
     "ID IS EQUAL TO 1"
  } 
  else if(id > 1){
    validateId(id)
  }
  else{
    "NO VALID RECORDS"
  }
})

def validateId(id:Int) : String = {
   if (id > 2) {
     "ID IS GREATER THAN 2"
   }
   else {
     "VALID RECORDS"
   }
 }

当我运行这段代码，我得到的任务不能序列例外。

有任何想法吗？谢谢。

Answer 1

udf被视为一个黑盒子，需要通过被序列化和反序列化因此不推荐使用udf的当你有内置功能的替代品的列。

调用udf功能与withColumn是好的，但你从造成这一问题的validateId函数中调用另一个函数udf。

我建议你不要使用udf功能，就好象你可以通过使用when内置功能实现的要求。

import org.apache.spark.sql.functions._
val formattedDF2 = df.withColumn("result_col", when($"id" === 1, lit("ID IS EQUAL TO 1")).otherwise(when($"id" > 2, lit("ID IS GREATER THAN 2")).otherwise(when($"id" > 1, lit("VALID RECORDS")).otherwise(lit("NO VALID RECORDS")))))