如何为任意类型的数组制作通用 UDF

【问题标题】：How to make generic UDF for Array of arbitrary types如何为任意类型的数组制作通用 UDF
【发布时间】：2018-04-12 13:26:36
【问题描述】：

假设我想制作一个 spark UDF 来反转结构数组的顺序。结构的具体类型应该无关紧要，所以我尝试了：

val reverseUDF = udf((s:Seq[_]) => s.reverse)

但这给了

java.lang.UnsupportedOperationException: Schema for type Any is not supported

我也尝试过使用泛型方法，强制类型泛型类型参数成为Product的子类型：

def reverse[T <: Product](s:Seq[T]) = {
  s.reverse
}

val reverseUDF = udf(reverse _)

这给出了：

scala.MatchError: Nothing (of class scala.reflect.internal.Types$TypeRef$$anon$6)

那么这甚至可能吗？

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

事实并非如此。 Spark 必须知道返回输出类型，并且无法使用 SQL 表达式来确定它。您必须为要使用的每种类型定义特定的udf，例如：

udf(reverse[(String, Int)] _)
udf(reverse[(String, Long, String)] _)

等等。然而，这些在实践中都没有用，因为你永远不会在你的 udf 中看到 Product 类型。结构类型始终编码为Row - Spark Sql UDF with complex input parameter。

如果您使用 Spark 2.3，您可以将任意 reverse 表示为：

import org.apache.spark.sql.Row
import org.apache.spark.sql.types.DataType

def reverse(schema: DataType) = udf(
  (xs: Seq[Row]) => xs.map(x => Row.fromSeq(x.toSeq.reverse)),
  schema
)

但您必须为每个实例提供架构：

【讨论】：