火花UDF过载

Question

我有一个要求，火花UDF必须超载，我知道UDF超载是不支持火花。因此，为了克服spark的这一限制，我尝试创建一个接受任何类型的UDF，它在UDF中找到实际的数据类型，并调用相应的方法进行计算并相应地返回值。这样做时，我得到一个错误

Exception in thread "main" java.lang.UnsupportedOperationException: Schema for type Any is not supported
    at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$schemaFor$1.apply(ScalaReflection.scala:789)
    at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$schemaFor$1.apply(ScalaReflection.scala:724)
    at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:56)
    at org.apache.spark.sql.catalyst.ScalaReflection$class.cleanUpReflectionObjects(ScalaReflection.scala:906)
    at org.apache.spark.sql.catalyst.ScalaReflection$.cleanUpReflectionObjects(ScalaReflection.scala:46)
    at org.apache.spark.sql.catalyst.ScalaReflection$.schemaFor(ScalaReflection.scala:723)
    at org.apache.spark.sql.catalyst.ScalaReflection$.schemaFor(ScalaReflection.scala:720)
    at org.apache.spark.sql.UDFRegistration.register(UDFRegistration.scala:213)
    at com.experian.spark_jobs.Test$.main(Test.scala:9)
    at com.experian.spark_jobs.Test.main(Test.scala)

以下是示例代码：

import org.apache.spark.sql.SparkSession

object Test {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[*]").appName("test").getOrCreate()
    spark.udf.register("testudf", testudf _)
    spark.sql("create temporary view testView as select testudf(1, 2) as a").show()
    spark.sql("select testudf(a, 5) from testView").show()

  }

  def testudf(a: Any, b: Any) = {
    if (a.isInstanceOf[Integer] && b.isInstanceOf[Integer]) {
      add(a.asInstanceOf[Integer], b.asInstanceOf[Integer])
    } else if (a.isInstanceOf[java.math.BigDecimal] && b.isInstanceOf[java.math.BigDecimal]) {
      add(a.asInstanceOf[java.math.BigDecimal], b.asInstanceOf[java.math.BigDecimal])
    }
  }

  def add(decimal: java.math.BigDecimal, decimal1: java.math.BigDecimal): java.math.BigDecimal = {
    decimal.add(decimal1)
  }

  def add(integer: Integer, integer1: Integer): Integer = {
    integer + integer1
  }
}

有可能使上述要求成为可能吗？如果没有，请建议我一个更好的方法。

注：Spark版本-2.4.0

Answer 1

使用Dataframe（未类型化）的问题是在编译时执行某种多态之类的操作非常痛苦。理想情况下，拥有列类型将允许使用特定的“添加函数”实现构建您的udfs，就像您使用Monides一样。但是Spark Dataframe API离这个世界非常遥远。使用数据集或无框架帮助很大。

在您的示例中，要在运行时检查类型，您将需要AnyRef而不是Any。那应该有用。