带有可变参数的 Spark UDF [英] Spark UDF with varargs

问题描述

是否是如文档中所示列出最多 22 个参数的唯一选项?

Is it an only option to list all the arguments up to 22 as shown in documentation?

https://spark.apache.org/docs/1.5.0/api/scala/index.html#org.apache.spark.sql.UDFRegistration

有人想出如何做类似的事情吗?

Anyone figured out how to do something similar to this?

sc.udf.register("func", (s: String*) => s......

(编写跳过空值的自定义 concat 函数，当时必须有 2 个参数)

(writing custom concat function that skips nulls, had to 2 arguments at the time)

谢谢

推荐答案

UDF 不支持可变参数*，但您可以传递使用 array 函数包装的任意数量的列:

UDFs don't support varargs* but you can pass an arbitrary number of columns wrapped using an array function:

import org.apache.spark.sql.functions.{udf, array, lit}

val myConcatFunc = (xs: Seq[Any], sep: String) => 
  xs.filter(_ != null).mkString(sep)

val myConcat = udf(myConcatFunc)

示例用法:

val  df = sc.parallelize(Seq(
  (null, "a", "b", "c"), ("d", null, null, "e")
)).toDF("x1", "x2", "x3", "x4")

val cols = array($"x1", $"x2", $"x3", $"x4")
val sep = lit("-")

df.select(myConcat(cols, sep).alias("concatenated")).show

// +------------+
// |concatenated|
// +------------+
// |       a-b-c|
// |         d-e|
// +------------+

使用原始 SQL:

df.registerTempTable("df")
sqlContext.udf.register("myConcat", myConcatFunc)

sqlContext.sql(
    "SELECT myConcat(array(x1, x2, x4), '.') AS concatenated FROM df"
).show

// +------------+
// |concatenated|
// +------------+
// |         a.c|
// |         d.e|
// +------------+

稍微复杂一点的方法是根本不使用 UDF，而是使用大致如下所示的内容组合 SQL 表达式:

A slightly more complicated approach is not use UDF at all and compose SQL expressions with something roughly like this:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.Column

def myConcatExpr(sep: String, cols: Column*) = regexp_replace(concat(
  cols.foldLeft(lit(""))(
    (acc, c) => when(c.isNotNull, concat(acc, c, lit(sep))).otherwise(acc)
  )
), s"($sep)?$$", "") 

df.select(
  myConcatExpr("-", $"x1", $"x2", $"x3", $"x4").alias("concatenated")
).show
// +------------+
// |concatenated|
// +------------+
// |       a-b-c|
// |         d-e|
// +------------+

但我怀疑这是否值得，除非您使用 PySpark.

but I doubt it is worth the effort unless you work with PySpark.

* 如果您使用 varargs 传递函数，它将从所有语法糖中剥离，并且生成的 UDF 将需要 ArrayType.例如:

* If you pass a function using varargs it will be stripped from all the syntactic sugar and resulting UDF will expect an ArrayType. For example:

def f(s: String*) = s.mkString
udf(f _)

将是以下类型:

UserDefinedFunction(<function1>,StringType,List(ArrayType(StringType,true)))

这篇关于带有可变参数的 Spark UDF的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！