spark不检查UDF / Column类型吗？

Question

考虑以下 Spark 2.1 代码：

val df = Seq("Raphael").toDF("name")
df.show()    

+-------+
|   name|
+-------+
|Raphael|
+-------+

val squareUDF = udf((d:Double) => Math.pow(d,2))

df.select(squareUDF($"name")).show

+---------+
|UDF(name)|
+---------+
|     null|
+---------+

为什么我得到null？我期待像 ClassCastException 这样的东西，因为我尝试将字符串映射到 Scala Double

"Raphael".asInstanceOf[Double]

java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Double

Answer 1

如果你检查执行计划很容易弄清楚：

scala> df.select(squareUDF($"name")).explain(true)
== Parsed Logical Plan ==
'Project [UDF('name) AS UDF(name)#51]
+- AnalysisBarrier Project [value#36 AS name#38]

== Analyzed Logical Plan ==
UDF(name): double
Project [if (isnull(cast(name#38 as double))) null else UDF(cast(name#38 as double)) AS UDF(name)#51]
+- Project [value#36 AS name#38]
   +- LocalRelation [value#36]

== Optimized Logical Plan ==
LocalRelation [UDF(name)#51]

== Physical Plan ==
LocalTableScan [UDF(name)#51]

如您所见，Spark 在应用 UDF 之前执行类型转换：

UDF(cast(name#38 as double))

并且 SQL 转换不会为类型兼容的转换抛出异常。如果无法进行实际转换，则该值未定义 (NULL)。如果类型不兼容：

Seq((1, ("Raphael", 42))).toDF("id", "name_number").select(squareUDF($"name_number"))
// org.apache.spark.sql.AnalysisException: cannot resolve 'UDF(name_number)' due to data type mismatch: argument 1 requires double type, however, '`name_number`' is of struct<_1:string,_2:int> type.;;
// 
// at org.apache...

你会得到一个例外。

如果类型不兼容

其余部分由：

if (isnull(cast(name#38 as double))) null 

由于值为null udf 永远不会被调用。