在 pyspark 的 Scala UDF 中使用默认参数值？

Question

我在 Scala 中定义了一个 UDF，其默认参数值如下：

package myUDFs

import org.apache.spark.sql.api.java.UDF3

class my_udf extends UDF3[Int, Int, Int, Int] {

  override def call(a: Int, b: Int, c: Int = 6): Int = {
    c*(a + b)
  }
}

然后我使用 build clean assembly 适当地构建它（如果需要可以提供更多构建细节）并提取 jar myUDFs-assembly-0.1.1.jar 并将其包含在 Python 的我的 Spark 配置中：

from pyspark.conf import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.types import IntType

spark_conf = SparkConf().setAll([
    ('spark.jars', 'myUDFs-assembly-0.1.1.jar')
])

spark = SparkSession.builder \
    .appName('my_app') \
    .config(conf = spark_conf) \
    .enableHiveSupport() \
    .getOrCreate()

spark.udf.registerJavaFunction(
    "my_udf", "myUDFs.my_udf", IntType()
)

但是，当我尝试利用默认值时，我被拒绝了：

spark.sql('select my_udf(1, 2)').collect()

AnalysisException: 'Invalid number of arguments for function my_udf. Expected: 3; Found: 2; line x pos y'

不可以有这样一个默认值的UDF吗？输出应该是 6*(1+2) = 18.

Answer 1

您在 spark sql 中调用函数时仅传递了两个参数。尝试传递三个参数

spark.sql('select my_udf(1, 2, 3 )').collect()

Answer 2

只要查看调用链，这里就没有机会识别默认参数。

Python registerJavaFunction invokes its JVM UDFRegistration.registerJava.
registerJava invokes matching register implementation.

其中，如果是UDF3，looks like this：

 * Register a deterministic Java UDF3 instance as user-defined function (UDF).
 * @since 1.3.0
 */
def register(name: String, f: UDF3[_, _, _, _], returnType: DataType): Unit = {
  val func = f.asInstanceOf[UDF3[Any, Any, Any, Any]].call(_: Any, _: Any, _: Any)
  def builder(e: Seq[Expression]) = if (e.length == 3) {
    ScalaUDF(func, returnType, e, e.map(_ => true), udfName = Some(name))
  } else {
    throw new AnalysisException("Invalid number of arguments for function " + name +
      ". Expected: 3; Found: " + e.length)
  }
  functionRegistry.createOrReplaceTempFunction(name, builder)
}

如您所见，builder 仅在实际调度调用之前验证提供的表达式是否与函数的 arity 匹配。

你可能会更幸运地实现一个中间 API ，它将处理默认参数并在幕后分派给 UDF。然而，这仅适用于 DataFrame API，因此它可能不符合您的需要。

在 pyspark 的 Scala UDF 中使用默认参数值？

Using default argument values in Scala UDF from pyspark?

scala

user-defined-functions

apache-spark

pyspark

pyspark-sql