Spark Sql udf具有可变数量的参数

Question

我想为Spark Sql连接函数。我已经写了一个udf

sqlContext.udf.register("CONCAT",(args:String*)=>{
 String out=""
 for(arg<-args)
  {
    out+=arg
  }
 out
})
sqlContext.sql("select col1,col2,CONCAT(col1,col2) from testtable")

，但这个udf不工作，我得到一个异常。如果我尝试使用固定数量的参数，那么它就可以工作。我使用spark 1.3.1和scala 2.10.5。

有人遇到过这个问题或知道解决方案吗?

Answer 1

您可以使用struct函数执行此操作，如下所示:

val myUDF = udf {
  (r: Row) => r.toSeq.map(...) // the "r" row contains your arguments
}
val df = ....
df.select(col("col1"), myUDF(struct(col("col2"), col("col3"), col("col4"), ...)))

Answer 2

如果您只想使用原始SQL连接列，则根本不需要自定义UDF。CONCAT功能已经存在:

val df = sc.parallelize(List(("a", "b", "c"))).toDF("x", "y", "z")
df.registerTempTable("df")
sqlContext.sql("SELECT CONCAT(x, y, z) AS xyz FROM df").show
// +---+
// |xyz|
// +---+
// |abc|
// +---+

从1.5.0开始，你可以直接使用concat/concat_ws函数:

import org.apache.spark.sql.functions.{concat, concat_ws}
df.select(concat_ws("-", $"x", $"y", $"z").alias("x-y-z")).show
// +-----+
// |x-y-z|
// +-----+
// |a-b-c|
// +-----+
df.select(concat($"x", $"y", $"z").alias("xyz")).show
// +---+
// |xyz|
// +---+
// |abc|
// +---+

参见Spark UDF with varargs