Spark UDF con vararg

Question

È una sola opzione per elencare tutti gli argomenti fino a 22 come mostrato nella documentazione?

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

Chiunque capito come fare qualcosa di simile a questo?

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

(scrittura funzione concat personalizzata che salta null, ha dovuto 2 argomenti al momento)

Grazie

Answer 1

UDF non supportano varargs *, ma è possibile passare un numero arbitrario di colonne avvolto utilizzando una funzione array:

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) 

un esempio d'uso:

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| 
// +------------+ 

Con SQL prime:

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| 
// +------------+ 

Un approccio un po 'più complicato non si usa UDF a tutti e comporre le espressioni SQL con qualcosa di meno così:

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| 
// +------------+ 

ma dubito ne vale la pena lo sforzo a meno che non si lavori con PySpark.

---

* Se si passa una funzione tramite varargs sarà spogliato da tutto lo zucchero sintattico e conseguente UDF si aspettano un ArrayType. Per esempio:

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

sarà di tipo:

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