Spark SQL 查询失败并出现 NullPointerException

Question

我正在尝试运行一个非常简单的 SQL 查询，涉及 join 和 orderby 子句，然后在最外面的 select stmt 中使用 UUID() 函数。查询失败

val query = spark.sql(" select name, uuid() as _iid from (select s.name from titanic s join titanic t on s.name = t.name order by name) ;")

出现以下错误

Exception in thread "main" java.lang.NullPointerException
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
at org.apache.spark.sql.execution.TakeOrderedAndProjectExec.$anonfun$executeCollect$2(limit.scala:207)
at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:237)
at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)
at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:198)
at scala.collection.TraversableLike.map(TraversableLike.scala:237)
at scala.collection.TraversableLike.map$(TraversableLike.scala:230)
at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:198)
at org.apache.spark.sql.execution.TakeOrderedAndProjectExec.executeCollect(limit.scala:207)
at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.$anonfun$executeCollect$1(AdaptiveSparkPlanExec.scala:338)
at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.withFinalPlanUpdate(AdaptiveSparkPlanExec.scala:366)
at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.executeCollect(AdaptiveSparkPlanExec.scala:338)
at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3715)
at org.apache.spark.sql.Dataset.$anonfun$head$1(Dataset.scala:2728)
at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3706)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:103)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:163)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:90)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3704)
at org.apache.spark.sql.Dataset.head(Dataset.scala:2728)
at org.apache.spark.sql.Dataset.take(Dataset.scala:2935)
at org.apache.spark.sql.Dataset.getRows(Dataset.scala:287)
at org.apache.spark.sql.Dataset.showString(Dataset.scala:326)
at org.apache.spark.sql.Dataset.show(Dataset.scala:808)
at org.apache.spark.sql.Dataset.show(Dataset.scala:785)
at hyperspace2.sparkPlan$.delayedEndpoint$hyperspace2$sparkPlan$1(sparkPlan.scala:14)
at hyperspace2.sparkPlan$delayedInit$body.apply(sparkPlan.scala:6)
at scala.Function0.apply$mcV$sp(Function0.scala:39)
at scala.Function0.apply$mcV$sp$(Function0.scala:39)
at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:17)
at scala.App.$anonfun$main$1$adapted(App.scala:80)
at scala.collection.immutable.List.foreach(List.scala:392)
at scala.App.main(App.scala:80)
at scala.App.main$(App.scala:78)
at hyperspace2.sparkPlan$.main(sparkPlan.scala:6)
at hyperspace2.sparkPlan.main(sparkPlan.scala)

注：

1. 如果我删除 order by 子句，那么它会产生正确的输出。
2. 如果我将 uuid() 更改为 monotonically_increasing_id() 那么它也可以工作。 （但这不是我的用例）

如何克服这个错误？为什么会出现这个错误，SQL 对我来说似乎合乎逻辑。

如果需要，我可以分享查询的逻辑计划。

Answer 1

这个有效，

  val input = Seq(
    (12, "abc"),
    (56, "def"),
    (90, "jkl")
  ).toDF("_id", "name")

  input.createOrReplaceTempView("titanic")

  println("spark.version:" + spark.version)

  spark
    .sql(
      " select name, uuid() as _iid from (select s.name from titanic s join titanic t on s.name = t.name order by name) ;"
    )
    .show(false)

  /*
    spark.version:3.3.0
    +----+------------------------------------+
    |name|_iid                                |
    +----+------------------------------------+
    |abc |c4a3d839-9f87-4fa8-b58f-a56be43a3425|
    |def |9f54d173-c4f2-47e6-bb83-9dbc6ad735cc|
    |jkl |6c1583ed-82e9-4e9c-9ce6-f511db84fdeb|
    +----+------------------------------------+
 */