Scala-Spark:不能使用UDF

Question

我在Spark（Scala）中使用UDF遇到问题。这是一个示例代码：

import org.apache.spark.sql.{SparkSession, DataFrame}
import org.apache.spark.sql.functions.{col, udf}

val spark = SparkSession.builder.appName("test") 
             .master("local[*]")
             .getOrCreate()
import spark.implicits._

def func(a: Array[Int]): Array[Int] = a
val funcUDF = udf((a: Array[Int]) => func(a))

var data = Seq(Array(1, 2, 3), Array(3, 4, 5), Array(6, 2, 4)).toDF("items")
data = data.withColumn("a", funcUDF(col("items")))
data.show()

我得到的错误与ClassCastException有关，表示不可能从＜code＞scala.collection.mutable强制转换。WrappedArray$ofReftoorg.apache.spark.sql.catalyst.expressions.ScalaUDF.$anonfun$f$2。我在下面添加了堆栈的一部分。如果有帮助，我正在使用https://community.cloud.databricks.com/.

Answer 1

问题是您的“items”列是WrappedArray类型(这是每个array like类型的Spark类型)。并且Array和WrappedArray之间没有隐式转换。所以我会建议使用Seq，因为WrappedArray是Seq的子类，但不是Array的子类。

这工作原理：

import org.apache.spark.sql.{SparkSession, DataFrame}
import org.apache.spark.sql.functions.{col, udf}

val spark = SparkSession.builder.appName("test") 
             .master("local[*]")
             .getOrCreate()
import spark.implicits._

def func(a: Array[Int]): Array[Int] = a
val funcUDF = udf((a: Seq[Int]) => func(a.toArray))

var data = Seq(Array(1, 2, 3), Array(3, 4, 5), Array(6, 2, 4)).toDF("items")
data = data.withColumn("a", funcUDF(col("items")))
data.show()