如何在 Spark 中创建 UDF 以支持自定义谓词

Question

我有一个数据帧，它有一个列表数据类型的字段，需要与交叉连接匹配，条件是如果列表中的任何元素存在于另一个列表中，那么这两个记录应被视为匹配。

例。

import org.apache.spark.sql.functions.udf

val df = sc.parallelize(Seq(("one", List(1,34,3)), ("one", List(1,2,3)), ("two", List(1))))
          .toDF("word", "count")
val lsEqual = (xs : (List[Int],List[Int])) => xs._1.find(xs._2.contains(_)).nonEmpty
 val equalList = udf(lsEqual)

但这给了我以下错误

val out =  df.joinWith(df,equalList(df("count"),df("count")),"cross")

java.lang.ClassCastException: $anonfun$1 cannot be cast to scala.Function2
at org.apache.spark.sql.catalyst.expressions.ScalaUDF.<init>(ScalaUDF.scala:97)
at org.apache.spark.sql.expressions.UserDefinedFunction.apply(UserDefinedFunction.scala:56)
... 50 elided

还有其他方法可以创建自定义谓词吗？

Answer 1

您的lsEqual函数定义似乎是错误的。 List、Seq、Array在Spark数据帧中被视为WrappedArray。您将两个columns传递给lsEqual函数，这应该是两个变量。

正确的方法应该是

val lsEqual = (xs1 : scala.collection.mutable.WrappedArray[Int], xs2 : scala.collection.mutable.WrappedArray[Int]) => xs1.find(xs2.contains(_)).nonEmpty

这绝对应该消除您面临的错误