spark udf 没有被调用答案

【问题标题】：spark udf not being calledspark udf 没有被调用
【发布时间】：2020-10-20 09:23:08
【问题描述】：

举个例子：

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions._

val testUdf: UserDefinedFunction = udf((a: String, b: String, c: Int) => { 
  val out = s"test1: $a $b $c"
  println(out)
  out
})

val testUdf2: UserDefinedFunction = udf((a: String, b: String, c: String) => { 
  val out = s"test2: $a $b $c"
  println(out)
  out
})

Seq(("hello", "world", null))
.toDF("a", "b", "c")
.withColumn("c", $"c" cast "Int")
.withColumn("test1", testUdf($"a", $"b", $"c"))
.withColumn("test2", testUdf2($"a", $"b", $"c"))
.show

testUdf 似乎没有被调用。没有错误，没有警告，它只是返回 null。

有没有办法检测这些静默故障？还有，这是怎么回事？

火花 2.4.4 斯卡拉 2.11

【问题讨论】：

标签： scala apache-spark apache-spark-sql user-defined-functions

【解决方案1】：

Scala 类型“Int”不允许空值。变量“c”类型可以改为“Integer”。

【讨论】：

感谢您的回答，+1。我真的很想了解为什么根本没有调用该函数，这在 yangzhongbao 的答案中有所体现

【解决方案2】：

我不知道是什么原因造成的。但我认为这很可能是因为隐式转换

代码1

    val spark = SparkSession.builder()
      .master("local")
      .appName("test")
      .getOrCreate()
    import spark.implicits._
    val testUdf: UserDefinedFunction = udf((a: String, b: String, c: Int) => {
      val out = s"test1: $a $b $c"
      println(out)
      out
    })
    
    Seq(("hello", "world", null))
      .toDF("a", "b", "c")
      .withColumn("test1", testUdf($"a", $"b", $"c"))
      .show

code2

    val spark = SparkSession.builder()
      .master("local")
      .appName("test")
      .getOrCreate()
    import spark.implicits._
    val testUdf: UserDefinedFunction = udf((a: String, b: String, c: String) => {
      val out = s"test1: $a $b $c"
      println(out)
      out
    })

    Seq(("hello", "world", null))
      .toDF("a", "b", "c")
      .withColumn("test1", testUdf($"a", $"b", $"c"))
      .show

code1 逻辑计划

code2 逻辑计划

【讨论】：

案例 1 的逻辑计划很好地解释了正在发生的事情。这有点出乎意料，但是，至少我知道发生了什么。谢谢！

【解决方案3】：

当您尝试强制转换为 null 时，您应该会遇到 scala.MatchError: scala.Null 错误，此外您对 UDF 的定义对我不起作用，因为我在尝试注册时收到了 java.lang.UnsupportedOperationException: Schema for type AnyRef is not supported。

这个呢：

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions._

def testUdf(a: String, b: String, c: Integer): String = { 
  val out = s"test1: $a $b $c"
  println(out)
  out
}

def testUdf2(a: String, b: String, c: String): String = { 
  val out = s"test2: $a $b $c"
  println(out)
  out
}

val yourTestUDF = udf(testUdf _)
val yourTestUDF2 = udf(testUdf2 _)

// spark.udf.register("yourTestUDF", yourTestUDF) // just in case you need it in SQL

spark.createDataFrame(Seq(("hello", "world", null.asInstanceOf[Integer])))
.toDF("a", "b", "c")
.withColumn("test1", yourTestUDF($"a", $"b", $"c"))
.withColumn("test2", yourTestUDF2($"a", $"b", $"c"))
.show(false)

输出：

test1: hello world null
test2: hello world null
+-----+-----+----+-----------------------+-----------------------+
|a    |b    |c   |test1                  |test2                  |
+-----+-----+----+-----------------------+-----------------------+
|hello|world|null|test1: hello world null|test2: hello world null|
+-----+-----+----+-----------------------+-----------------------+

【讨论】：