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 有关，说不可能从 scala.collection.mutable.WrappedArray$ofRef 转换为 org.apache.spark.sql.catalyst.expressions.ScalaUDF.$anonfun$f$2。我在下面添加了堆栈的一部分。如果有帮助，我正在使用https://community.cloud.databricks.com/。

引起：java.lang.ClassCastException： scala.collection.mutable.WrappedArray$ofRef 不能转换为 [I at org.apache.spark.sql.catalyst.expressions.ScalaUDF.$anonfun$f$2(ScalaUDF.scala:155) 在 org.apache.spark.sql.catalyst.expressions.ScalaUDF.eval(ScalaUDF.scala:1125) 在 org.apache.spark.sql.catalyst.expressions.Alias.eval(namedExpressions.scala:156) 在 org.apache.spark.sql.catalyst.expressions.InterpretedMutableProjection.apply(InterpretedMutableProjection.scala:83) 在 org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$$anonfun$apply$15.$anonfun$applyOrElse$70(Optimizer.scala:1557) 在 scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238) 在 scala.collection.immutable.List.foreach(List.scala:392) 在 scala.collection.TraversableLike.map(TraversableLike.scala:238) 在 scala.collection.TraversableLike.map$(TraversableLike.scala:231) 在 scala.collection.immutable.List.map(List.scala:298) 在 org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$$anonfun$apply$15.applyOrElse(Optimizer.scala:1557) 在 org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$$anonfun$apply$15.applyOrElse(Optimizer.scala:1552) 在 org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDown$1(TreeNode.scala:322) 在 org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:80) 在 org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:322) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown（AnalysisHelper.scala:153） 在 org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown$(AnalysisHelper.scala:151) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDown$3(TreeNode.scala:327) 在 org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$mapChildren$1(TreeNode.scala:412) 在 org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:250) 在 org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:410) 在 org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:363) 在 org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:327) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown（AnalysisHelper.scala:153） 在 org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown$(AnalysisHelper.scala:151) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDown$3(TreeNode.scala:327) 在 org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$mapChildren$1(TreeNode.scala:412) 在 org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:250) 在 org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:410) 在 org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:363) 在 org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:327) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown（AnalysisHelper.scala:153） 在 org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown$(AnalysisHelper.scala:151) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDown$3(TreeNode.scala:327) 在 org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$mapChildren$1(TreeNode.scala:412) 在 org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:250) 在 org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:410) 在 org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:363) 在 org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:327) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown（AnalysisHelper.scala:153） 在 org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown$(AnalysisHelper.scala:151) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29) 在 org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:311) 在 org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$.apply(Optimizer.scala:1552) 在 org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$.apply(Optimizer.scala:1551) 在 org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$2(RuleExecutor.scala:152) 在 scala.collection.IndexedSeqOptimized.foldLeft(IndexedSeqOptimized.scala:60) 在 scala.collection.IndexedSeqOptimized.foldLeft$(IndexedSeqOptimized.scala:68) 在 scala.collection.mutable.WrappedArray.foldLeft(WrappedArray.scala:38) 在 org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1(RuleExecutor.scala:149) 在 org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1$adapted(RuleExecutor.scala:141) 在 scala.collection.immutable.List.foreach(List.scala:392) 在 org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:141) 在 org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$executeAndTrack$1(RuleExecutor.scala:119) 在 org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:88) 在 org.apache.spark.sql.catalyst.rules.RuleExecutor.executeAndTrack(RuleExecutor.scala:119) 在 org.apache.spark.sql.execution.QueryExecution.$anonfun$optimizedPlan$1(QueryExecution.scala:107) 在 org.apache.spark.sql.catalyst.QueryPlanningTracker.measurePhase(QueryPlanningTracker.scala:111) 在 org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$1(QueryExecution.scala:171) 在 org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:836) 在 org.apache.spark.sql.execution.QueryExecution.executePhase(QueryExecution.scala:171) 在 org.apache.spark.sql.execution.QueryExecution.optimizedPlan$lzycompute(QueryExecution.scala:104) 在 org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:104) 在 org.apache.spark.sql.execution.QueryExecution.$anonfun$writePlans$4(QueryExecution.scala:246) 在 org.apache.spark.sql.catalyst.plans.QueryPlan$.append(QueryPlan.scala:466) 在 org.apache.spark.sql.execution.QueryExecution.org$apache$spark$sql$execution$QueryExecution$$writePlans(QueryExecution.scala:246) 在 org.apache.spark.sql.execution.QueryExecution.toString(QueryExecution.scala:256) 在 org.apache.spark.sql.execution.SQLExecution$.$anonfun$withCustomExecutionEnv$5(SQLExecution.scala:109) 在 org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:249) 在 org.apache.spark.sql.execution.SQLExecution$.$anonfun$withCustomExecutionEnv$1(SQLExecution.scala:101) 在 org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:836) 在 org.apache.spark.sql.execution.SQLExecution$.withCustomExecutionEnv(SQLExecution.scala:77) 在 org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:199) 在 org.apache.spark.sql.Dataset.withAction(Dataset.scala:3700) 在 org.apache.spark.sql.Dataset.head(Dataset.scala:27​​11) 在 org.apache.spark.sql.Dataset.take(Dataset.scala:2918) 在 org.apache.spark.sql.Dataset.getRows(Dataset.scala:305) 在 org.apache.spark.sql.Dataset.showString(Dataset.scala:342) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:838) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:797) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:806) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:14) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:164) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:166) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:168) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:170) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:172) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:174) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:176) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw$$iw.(command-1114467142343660:178) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw$$iw.(command-1114467142343660:180) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw$$iw.(command-1114467142343660:182) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw$$iw.(command-1114467142343660:184) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$ $iw.(command-1114467142343660:186) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw。 （命令 1114467142343660:188） 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(命令- 1114467142343660:190) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:192 ) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:194) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:196) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:198) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:200) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:202) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:204) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:206) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw$$iw.(command-1114467142343660:208) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw$$iw.(command-1114467142343660:210) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw$$iw.(command-1114467142343660:212) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw$$iw.(command-1114467142343660:214) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$$iw.(command-1114467142343660:216) 在 lineedcf33d032244134ad784ac9de826d3b265.$read.(command-1114467142343660:218) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$.(command-1114467142343660:222) 在 lineedcf33d032244134ad784ac9de826d3b265.$read$.(command-1114467142343660) 在 lineedcf33d032244134ad784ac9de826d3b265.$eval$.$print$lzycompute(:7) 在 lineedcf33d032244134ad784ac9de826d3b265.$eval$.$print(:6) 在 lineedcf33d032244134ad784ac9de826d3b265.$eval.$print() 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:745) 在 scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:1021) 在 scala.tools.nsc.interpreter.IMain.$anonfun$interpret$1(IMain.scala:574) 在 scala.reflect.internal.util.ScalaClassLoader.asContext(ScalaClassLoader.scala:41) 在 scala.reflect.internal.util.ScalaClassLoader.asContext$(ScalaClassLoader.scala:37) 在 scala.reflect.internal.util.AbstractFileClassLoader.asContext(AbstractFileClassLoader.scala:41) 在 scala.tools.nsc.interpreter.IMain.loadAndRunReq$1(IMain.scala:573) 在 scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:600) 在 scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:570) 在 com.databricks.backend.daemon.driver.DriverILoop.execute（DriverILoop.scala:219） 在 com.databricks.backend.daemon.driver.ScalaDriverLocal.$anonfun$repl$1(ScalaDriverLocal.scala:204) 在 scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) 在 com.databricks.backend.daemon.driver.DriverLocal$TrapExitInternal$.trapExit(DriverLocal.scala:773) 在 com.databricks.backend.daemon.driver.DriverLocal$TrapExit$.apply(DriverLocal.scala:726) 在 com.databricks.backend.daemon.driver.ScalaDriverLocal.repl(ScalaDriverLocal.scala:204) 在 com.databricks.backend.daemon.driver.DriverLocal.$anonfun$execute$10(DriverLocal.scala:431) 在 com.databricks.logging.UsageLogging.$anonfun$withAttributionContext$1(UsageLogging.scala:239) 在 scala.util.DynamicVariable.withValue(DynamicVariable.scala:62) 在 com.databricks.logging.UsageLogging.withAttributionContext(UsageLogging.scala:234) 在 com.databricks.logging.UsageLogging.withAttributionContext$(UsageLogging.scala:231) 在 com.databricks.backend.daemon.driver.DriverLocal.withAttributionContext(DriverLocal.scala:48) 在 com.databricks.logging.UsageLogging.withAttributionTags（UsageLogging.scala:27​​6） 在 com.databricks.logging.UsageLogging.withAttributionTags$(UsageLogging.scala:269) 在 com.databricks.backend.daemon.driver.DriverLocal.withAttributionTags（DriverLocal.scala:48） 在 com.databricks.backend.daemon.driver.DriverLocal.execute（DriverLocal.scala:408） 在 com.databricks.backend.daemon.driver.DriverWrapper.$anonfun$tryExecutingCommand$1(DriverWrapper.scala:653) 在 scala.util.Try$.apply(Try.scala:213) 在 com.databricks.backend.daemon.driver.DriverWrapper.tryExecutingCommand(DriverWrapper.scala:645) 在 com.databricks.backend.daemon.driver.DriverWrapper.getCommandOutputAndError（DriverWrapper.scala:486） 在 com.databricks.backend.daemon.driver.DriverWrapper.executeCommand（DriverWrapper.scala:598） 在 com.databricks.backend.daemon.driver.DriverWrapper.runInnerLoop（DriverWrapper.scala:391） 在 com.databricks.backend.daemon.driver.DriverWrapper.runInner（DriverWrapper.scala:337） 在 com.databricks.backend.daemon.driver.DriverWrapper.run（DriverWrapper.scala:219） 在 java.lang.Thread.run(Thread.java:748)

Answer 1

问题是您的“项目”列是 WrappedArray 类型（这是每个数组类型的 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()