Hinzufügen neuer Spalten basierend auf Aggregation auf vorhandene Spalte in Spark DataFrame mithilfe von Scala

Question

Ich habe einen DataFrame wie unten. Ich muss eine neue Spalte basierend auf vorhandenen Spalten erstellen.

col1 col2 
a  1 
a  2 
b  1 
c  1 
d  1 
d  2 

Output Data Frame sehen aus wie dieses

col1 col2 col3 col4 
a  1 1  2 
a  2 1  2 
b  1 0  1 
c  1 0  1 
d  1 1  2 
d  2 1  2 

Die Logik, die ich verwendet habe col3 zu finden ist wenn Graf von col1> 1 und col4 max-Wert von col2 ist.

Ich bin vertraut mit, wie man es in SQL macht. Mit Dataframe DSL ist es jedoch schwierig, eine Lösung zu finden. Jede Hilfe wäre willkommen. Danke

Answer 1

groupBy col1 und Aggregat zu erhalten count und max. Dann können Sie es wieder mit Original-Datenrahmen join Ihre gewünschte Ergebnis

val df2 = df1.groupBy("col1").agg(count() as col3, max("col2") as col4) 

val df3 = df1.join(df2, "col1") 

Answer 2

Funken df erhalten hat Eigenschaft withColumn genannt können Sie fügen Sie so viele abgeleitete Spalten, wie Sie wollen. Die Spalte wird jedoch nicht zum vorhandenen DF hinzugefügt, sondern erstellt einen neuen DF mit einer hinzugefügten Spalte.

z.B. Hinzufügen eines statischen Datum auf die Daten

val myFormattedData = myData.withColumn("batchdate",addBatchDate(myData("batchdate"))) 
val addBatchDate = udf { (BatchDate: String) => "20160101" } 

Answer 3

col3 Hinzufügen Sie withcolumn verwenden + wenn/sonst:

val df2 = df.withColumn("col3",when($"col2" > 1, 1).otherwise(0)) 

Hinzufügen COL4 die groupBy/max + beitreten bereits erwähnt sollte die Arbeit machen:

val df3 = df2.join(df.groupBy("col1").max("col2"), "col1")