Spark Pivot eine Spalte, aber andere intakt halten

Question

Wie kann ich die folgenden Werte angeben, aber wie kann ich die maximalen Werte pivotieren, aber die Summe der Spiele zusammenfassen?

from pyspark import SparkContext 
from pyspark.sql import HiveContext 
from pyspark.sql import functions as F 
from pyspark.sql import Window 

df = sqlContext.createDataFrame([ 
    ("u1", "g1", 10, 0, 1), 
    ("u1", "g3", 2, 2, 1), 
    ("u1", "g3", 5, 3, 1), 
    ("u1", "g4", 5, 4, 1), 
    ("u2", "g2", 1, 1, 1), 
], ["UserID", "GameID", "Score", "Time", "Plays"]) 

gewünschte Ausgabe

+------+-------------+-------------+-----+ 
|UserID|MaxScoreGame1|MaxScoreGame2|Plays| 
+------+-------------+-------------+-----+ 
| u1|   10|   5| 4| 
| u2|   1|   null| 1| 
+------+-------------+-------------+-----+ 

gab ich eine Lösung unten, aber ich bin der Hoffnung, mit zu vermeiden verbinden.

Answer 1

Ich glaube nicht, es ist eine echte Verbesserung, aber können Sie Gesamtzahl der Stücke

... 
.select(
    F.col("*"), 
    F.row_number().over(rowNumberWindow).alias("GameNumber"), 
    F.sum("Plays").over(rowNumberWindow.orderBy()).alias("total_plays") 
) 
... 

und verwenden Sie es später als sekundäre Gruppierungsspalte für pivot hinzufügen:

... 
.groupBy("UserID", "total_plays") 
.pivot("GameCol", ["MaxScoreGame1", "MaxScoreGame2"]) 
.agg(F.max("Score")) 
... 

Answer 2

Hier ist eine Lösung beitreten verwenden, die ich hoffe, ich zu vermeiden:

Sum Datenrahmen

df_sum = df.groupBy("UserID").agg(F.sum("Plays").alias("Plays")).alias("df_sum") 
df_sum.show() 

+------+-----+ 
|UserID|Plays| 
+------+-----+ 
| u1| 4| 
| u2| 1| 
+------+-----+ 

rowNumberWindow = Window.partitionBy("UserID").orderBy(F.col("Time")) 

Pivot Datenrahmen

rowNumberWindow = Window.partitionBy("UserID").orderBy(F.col("Time")) 

df_piv = (df 
     .groupBy("UserID", "GameID") 
     .agg(F.sum("Plays").alias("Plays"), 
       F.max("Score").alias("Score"), 
       F.min("Time").alias("Time")) 
     .select(F.col("*"), 
       F.row_number().over(rowNumberWindow).alias("GameNumber")) 
     .filter(F.col("GameNumber") <= F.lit(2)) 
     .withColumn("GameCol", F.concat(F.lit("MaxScoreGame"), F.col("GameNumber"))) 
     .groupBy("UserID") 
     .pivot("GameCol", ["MaxScoreGame1", "MaxScoreGame2"]) 
     .agg(F.max("Score")) 
     ).alias("df_piv") 

df_piv.show() 

+------+-------------+-------------+ 
|UserID|MaxScoreGame1|MaxScoreGame2| 
+------+-------------+-------------+ 
| u1|   10|   5| 
| u2|   1|   null| 
+------+-------------+-------------+ 

Registriert Datenrahmen

df_joined = df_sum.join(df_piv, F.col("df_sum.UserID") == F.col("df_piv.UserID")) 

df_joined.show() 

+------+-----+------+-------------+-------------+ 
|UserID|Plays|UserID|MaxScoreGame1|MaxScoreGame2| 
+------+-----+------+-------------+-------------+ 
| u1| 4| u1|   10|   5| 
| u2| 1| u2|   1|   null| 
+------+-----+------+-------------+-------------+