创建Spark exists用户定义函数，其工作方式类似于Scala Array#exists函数

Question

Scala有一个Array#exists函数，工作原理是这样的:

Array(1, 2, 5).exists(_ % 2 == 0) // true

我想创建一个类似的火花存在功能。假设我们有以下< code>sourceDF:

+---------+
|     nums|
+---------+
|[1, 4, 9]|
|[1, 3, 5]|
+---------+

我希望能够写出这样的东西:

val actualDF = sourceDF.withColumn(
  "nums_has_even",
  exists(col("nums"), (x: Int) => x % 2 == 0)
)

这是我写的代码：

def existsInt(arr: Array[Int], f: (Int => Boolean)): Boolean = {
  arr.exists(f(_))
}

val exists = udf[Boolean, Array[Int], (Int => Boolean)](existsInt)

我理解为什么我的代码不工作。UDF需要列参数，匿名函数不是列对象。在＜code＞lit＜code＞中包装匿名函数不起作用：

exists(col("nums"), lit((x: Int) => x % 2 == 0)) // doesn't work

如何使此代码正常工作？

Answer 1

你很接近：

def existsInt(f: (Int => Boolean)) = udf {
  (arr: Seq[Int]) => arr.exists(f(_))  // Not Array!
}

用法：

existsInt((x: Int) => x % 2 == 0)(col("nums"))

您甚至可以:

scala.reflect.runtime.universe._

def exists[T : TypeTag](f: (T => Boolean)) = udf[Boolean, Seq[T]]{
  (arr: Seq[T]) => arr.exists(f(_))  // Not Array!
}

exists[Int]((x: Int) => x % 2 == 0).apply(col("nums"))