Использование значений аргументов по умолчанию в Scala UDF из pyspark?

Question

У меня определен UDF в Scala со значением аргумента по умолчанию, например:

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 и включите это в мою конфигурацию Spark в Python:

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:'Неверное количество аргументов для функции my_udf.Ожидаемое: 3;Найдено: 2;line x pos y '

Разве невозможно иметь UDF со значением по умолчанию, подобным этому?Выход должен быть 6*(1+2) = 18.

Answer 1

Просто глядя на цепочку вызовов, вы не сможете распознать аргумент по умолчанию.

• Python registerJavaFunction вызывает JVM UDFRegistration.registerJava.
• registerJava вызывает сопоставление register реализации .

• Что в случае UDF3, выглядит следующим образом :

   * 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, поэтому оно может не соответствовать вашим потребностям.

Answer 2

Вы передаете только два аргумента при вызове функции в spark sql.Попробуйте передать три аргумента

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