Funken-UDF, die unbekannte Anzahl von Spalten einnimmt

Question

Ich habe eine Liste von Spark-Datenrahmen mit verschiedenen Schemas. Beispiel:

list_df = [df1, df2, df3, df4] 
# df1.columns = ['a', 'b'] 
# df2.columns = ['a', 'b', 'c'] 
# df3.columns = ['a', 'b', 'c', 'd'] 
# df4.columns = ['a', 'b', 'c', 'd', 'e'] 

Nun möchte ich eine einzige UDF schreiben, die Lage in dieser Liste von Datenrahmen mit einer unterschiedlichen Anzahl von Spalten zu arbeiten.

Es gibt einen früheren Beitrag, wie man es mit scala: Spark UDF with varargs, wo das udf in einem Array von Spalten nimmt.

Aber es scheint, dass der Ansatz für Python nicht funktioniert. Irgendwelche Vorschläge?

Danke.

Answer 1

Eigentlich funktioniert dieser Ansatz nur gut in Python:

from pyspark.sql.functions import array, udf 

df = sc.parallelize([("a", "b", "c", "d")]).toDF() 

f = udf(lambda xs: "+".join(xs)) 

df.select(f("_1")).show() 
## +------------+ 
## |<lambda>(_1)| 
## +------------+ 
## |   a| 
## +------------+ 

df.select(f(array("_1", "_2"))).show() 
## +-----------------------+ 
## |<lambda>(array(_1, _2))| 
## +-----------------------+ 
## |     a+b| 
## +-----------------------+ 

df.select(f(array("_1", "_2", "_3"))).show() 
## +---------------------------+ 
## |<lambda>(array(_1, _2, _3))| 
## +---------------------------+ 
## |      a+b+c| 
## +---------------------------+ 

Da Python UDF nicht die gleiche Art von Unternehmen wie ihr Scala Pendant ist nicht durch die Art und Anzahl der Eingabeargumente Sie auch args verwenden eingeschränkt :

, um die Umbrucheingabe mit array zu vermeiden.

Sie auch struct als Alternative Wrapper verwenden können den Zugriff auf den Spaltennamen zu erhalten:

h = udf(lambda row: "+".join(row.asDict().keys())) 

df.select(h(struct("_1", "_2", "_3"))).show() 
## +----------------------------+ 
## |<lambda>(struct(_1, _2, _3))| 
## +----------------------------+ 
## |     _1+_3+_2| 
## +----------------------------+