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

Question

Scala 有一个 Array#exists 函数，其工作方式如下：

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

我想创建一个工作方式类似的 Spark exists 函数。假设我们有以下 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 需要列参数，匿名函数不是 Column 对象。在 lit 中包装匿名函数无效：

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"))

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

Creating a Spark exists user defined function that works like the Scala Array#exists function

scala

user-defined-functions

apache-spark

apache-spark-sql