Spalte Summe als neue Spalte in PySpark Datenrahmen hinzufügen

Question

Ich benutze PySpark und ich habe einen Spark-Datenrahmen mit einer Reihe von numerischen Spalten. Ich möchte eine Spalte hinzufügen, die die Summe aller anderen Spalten ist.

Angenommen mein Datenframe hatte Spalten "a", "b" und "c". Ich weiß, ich kann dies tun:

df.withColumn('total_col', df.a + df.b + df.c) 

Das Problem ist, dass ich möchte nicht jede Spalte einzeln tippen und fügen Sie sie, vor allem, wenn ich viele Spalten haben. Ich möchte dies automatisch tun oder eine Liste von Spaltennamen angeben, die ich hinzufügen möchte. Gibt es einen anderen Weg, dies zu tun?

Answer 1

Dies war nicht offensichtlich. Ich sehe keine zeilenbasierte Summe der in der Spark-Dataframe-API definierten Spalten.

Version 2

Dies kann auf eine recht einfache Art und Weise erfolgen: alle Spaltennamen in der Spark-Datenrahmen von pyspark als eine Liste von Zeichenketten geben

newdf = df.withColumn('total', sum(df[col] for col in df.columns)) 

df.columns geliefert. Für eine andere Summe können Sie stattdessen eine andere Liste mit Spaltennamen angeben.

Ich habe dies nicht als meine erste Lösung versucht, weil ich nicht sicher war, wie es sich verhalten würde. Aber es funktioniert.

Version 1

Dies ist zu kompliziert, aber wie gut funktioniert.

Sie können dies tun:

1. Verwendung df.columns eine Liste der Namen der Spalten zu bekommen
2. Verwendung, deren Namen Liste eine Liste der Spalten zu machen
3. Pass, der auf etwas Liste, die wird die Spalte überladene Add-Funktion in einem reduce die mit python

fold-type functional manner aufrufen, einige Kenntnisse, wie Arbeiten Betreiber Überlastung und die pyspark Code für Spalten here, die wird: reduzieren,

def column_add(a,b): 
    return a.__add__(b) 

newdf = df.withColumn('total_col', 
     reduce(column_add, (df[col] for col in df.columns))) 

Hinweis: Dies ist ein Python kein Funke RDD reduzieren und die Klammer Begriff im zweiten Parameter zu reduzieren, erfordert die Klammer, weil es ein Listengenerator Ausdruck ist.

Getestet, funktioniert!

$ pyspark 
>>> df = sc.parallelize([{'a': 1, 'b':2, 'c':3}, {'a':8, 'b':5, 'c':6}, {'a':3, 'b':1, 'c':0}]).toDF().cache() 
>>> df 
DataFrame[a: bigint, b: bigint, c: bigint] 
>>> df.columns 
['a', 'b', 'c'] 
>>> def column_add(a,b): 
...  return a.__add__(b) 
... 
>>> df.withColumn('total', reduce(column_add, (df[col] for col in df.columns))).collect() 
[Row(a=1, b=2, c=3, total=6), Row(a=8, b=5, c=6, total=19), Row(a=3, b=1, c=0, total=4)]