Как передать целую строку в UDF - фильтр Spark DataFrame

Я пишу функцию фильтра для сложного набора данных JSON с множеством внутренних структур. Передача отдельных столбцов слишком громоздка.

Итак, я объявил следующий UDF:

val records:DataFrame = = sqlContext.jsonFile("...")
def myFilterFunction(r:Row):Boolean=???
sqlc.udf.register("myFilter", (r:Row)=>myFilterFunction(r))

Интуитивно я думаю, что он будет работать следующим образом:

records.filter("myFilter(*)=true")

Каков фактический синтаксис?

Ответ 1

Вы должны использовать функцию struct() для построения строки при вызове функции, выполните следующие действия.

Импорт строки,

import org.apache.spark.sql._

Определите UDF

def myFilterFunction(r:Row) = {r.get(0)==r.get(1)}

Зарегистрируйте UDF

sqlContext.udf.register("myFilterFunction", myFilterFunction _)

Создайте файл данных

val records = sqlContext.createDataFrame(Seq(("sachin", "sachin"), ("aggarwal", "aggarwal1"))).toDF("text", "text2")

Используйте UDF

records.filter(callUdf("myFilterFunction",struct($"text",$"text2"))).show

Когда u хочет, чтобы все столбцы были переданы в UDF.

records.filter(callUdf("myFilterFunction",struct(records.columns.map(records(_)) : _*))).show

Результат:

+------+------+
|  text| text2|
+------+------+
|sachin|sachin|
+------+------+