Spark UDF как параметр функции, UDF не находится в области функций

Question

У меня есть несколько UDF, которые я хотел бы передать как аргумент функции вместе с кадрами данных.

Одним из способов сделать это может быть создание UDF внутри функции, но это создаст и уничтожит несколько экземпляров UDF без повторного использования, что может быть не лучшим способом подойти к этой проблеме.

Вот пример кусок кода -

val lkpUDF = udf{(i: Int) => if (i > 0) 1 else 0} 

val df = inputDF1 
    .withColumn("new_col", lkpUDF(col("c1"))) 
val df2 = inputDF2. 
    .withColumn("new_col", lkpUDF(col("c1"))) 

Вместо того, чтобы делать выше, я бы в идеале хотели бы сделать что-то вроде этого -

val lkpUDF = udf{(i: Int) => if (i > 0) 1 else 0} 

def appendCols(df: DataFrame, lkpUDF: ?): DataFrame = { 

    df 
     .withColumn("new_col", lkpUDF(col("c1"))) 

    } 
val df = appendCols(inputDF, lkpUDF) 

выше UDF довольно проста, но в моем случае он может вернуть примитивный тип или определяемый пользователем тип класса case. Любые мысли/указатели были бы высоко оценены. Благодарю.

Answer 1

Ваша функция с соответствующей подписью должен быть таким:

import org.apache.spark.sql.UserDefinedFunction 

def appendCols(df: DataFrame, func: UserDefinedFunction): DataFrame = { 
    df.withColumn("new_col", func(col("col1"))) 
} 

Скала РЕПЛ является весьма полезным при возвращении тип значения инициализированы.

scala> val lkpUDF = udf{(i: Int) => if (i > 0) 1 else 0} 
lkpUDF: org.apache.spark.sql.UserDefinedFunction = UserDefinedFunction(<function1>,IntegerType,List(IntegerType)) 

Кроме того, если подпись функции, которую вы передаете в udf обертки состоит из типа Any возврата (который будет в случае, если функция может возвращать либо примитивную или определенный класс случая пользователя), UDF не сможет скомпилироваться с таким исключением:

java.lang.UnsupportedOperationException: Schema for type Any is not supported