如何将整行传递给UDF-Spark DataFrame过滤器

Question

我正在为具有大量内部结构的复杂JSON数据集编写过滤函数。传递单个列太麻烦了。

因此，我声明了以下UDF：

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

直觉上，我认为它会这样工作：

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

实际的语法是什么？

Answer 1

除了第一个答案。当我们希望将所有列传递给UDF时，我们可以使用

 struct("*")

Answer 2

scala> inputDF
res40: org.apache.spark.sql.DataFrame = [email: string, first_name: string ... 3 more fields]

scala> inputDF.printSchema
root
 |-- email: string (nullable = true)
 |-- first_name: string (nullable = true)
 |-- gender: string (nullable = true)
 |-- id: long (nullable = true)
 |-- last_name: string (nullable = true)

现在，我想根据性别字段过滤行。我可以通过使用. filter（$"性别" === "男性"）来实现这一点，但我想使用. filter（函数）。

因此，定义了我的匿名函数

val isMaleRow = (r:Row) => {r.getAs("gender") == "Male"}

val isFemaleRow = (r:Row) => { r.getAs("gender") == "Female" }

inputDF.filter(isMaleRow).show()

inputDF.filter(isFemaleRow).show()

我觉得这个需求可以用更好的方式完成，即不用声明为UDF并调用它。

Answer 3

在调用函数时，必须使用＜code＞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|
+------+------+