Funken UDF mit Varargs

Question

Ist es eine einzige Option, um alle Argumente bis zu 22, wie in der Dokumentation gezeigt, aufzulisten?

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

Wer herausgefunden, wie dies etwas ähnliches zu tun?

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

(Schreiben von benutzerdefinierten concat-Funktion, die NULL-Werte überspringt, musste zwei Argumente zu der Zeit)

Dank

Answer 1

UDF nicht unterstützen varargs *, aber Sie können eine beliebige Anzahl von Spalten passieren gewickelt Verwendung einer array Funktion:

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) 

ein Beispiel Nutzung:

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

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

Ein etwas komplizierter Ansatz nicht verwenden UDF und SQL-Ausdrücke mit etwas komponieren grob wie folgt aus:

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

aber ich bezweifle es wert ist die Mühe, wenn Sie nicht mit PySpark arbeiten.

---

* Wenn Sie eine Funktion varargs passieren wird es von allen syntaktischen Zucker abgezogen werden und die daraus resultierende UDF wird eine ArrayType erwarten. Zum Beispiel:

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

wird der Typ sein:

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