Wie wähle ich effizient die erste Zeile jeder Gruppe mit dem höchsten Wert in einem Spark-DataFrame aus?

So wählen Sie die Homepage jeder Gruppe aus

Das Ziel besteht darin, die erste Seite mit dem höchsten Wert aus jeder Gruppe des DataFrame zu extrahieren. Dies ist häufig der Fall, wenn Datentrends analysiert oder die Top-Player in einer bestimmten Kategorie identifiziert werden. Um dies zu erreichen, können mehrere Methoden verwendet werden:

Fensterfunktion:

Fensterfunktionen bieten eine Möglichkeit, Berechnungen innerhalb einer Gruppe durchzuführen. In diesem Fall können wir die Funktion row_number() verwenden, um jeder Zeile basierend auf der angegebenen Reihenfolge eine Sequenznummer zuzuweisen. Die Homepage kann dann identifiziert werden, indem das Ranking auf 1 gefiltert wird.

import org.apache.spark.sql.functions.{row_number, max, broadcast}
import org.apache.spark.sql.expressions.Window

val df = sc.parallelize(Seq(
  (0,"cat26",30.9), (0,"cat13",22.1), (0,"cat95",19.6), (0,"cat105",1.3),
  (1,"cat67",28.5), (1,"cat4",26.8), (1,"cat13",12.6), (1,"cat23",5.3),
  (2,"cat56",39.6), (2,"cat40",29.7), (2,"cat187",27.9), (2,"cat68",9.8),
  (3,"cat8",35.6))).toDF("Hour", "Category", "TotalValue")

val w = Window.partitionBy($"hour").orderBy($"TotalValue".desc)

val dfTop = df.withColumn("rn", row_number.over(w)).where($"rn" === 1).drop("rn")

dfTop.show
// +----+--------+----------+
// |Hour|Category|TotalValue|
// +----+--------+----------+
// |   0|   cat26|      30.9|
// |   1|   cat67|      28.5|
// |   2|   cat56|      39.6|
// |   3|    cat8|      35.6|
// +----+--------+----------+

Einfacher SQL-Post-Aggregation-Join:

Alternativ können wir SQL verwenden, um die Aggregation durchzuführen und dann die Ergebnisse mit dem ursprünglichen DataFrame zu verketten, um die erste Seite jeder Gruppe zu extrahieren.

val dfMax = df.groupBy($"hour".as("max_hour")).agg(max($"TotalValue").as("max_value"))

val dfTopByJoin = df.join(broadcast(dfMax),
    ($"hour" === $"max_hour") && ($"TotalValue" === $"max_value"))
  .drop("max_hour")
  .drop("max_value")

dfTopByJoin.show

// +----+--------+----------+
// |Hour|Category|TotalValue|
// +----+--------+----------+
// |   0|   cat26|      30.9|
// |   1|   cat67|      28.5|
// |   2|   cat56|      39.6|
// |   3|    cat8|      35.6|
// +----+--------+----------+

Struktursortierung verwenden:

Eine gute Möglichkeit, die gleichen Ergebnisse zu erzielen, ohne Fensterfunktionen oder Joins zu verwenden, besteht darin, die Daten basierend auf einer Struktur mit Werten und Kategorien zu sortieren. Der Maximalwert dieser Struktur gibt dann die gewünschte erste Seite für jede Gruppe zurück.

val dfTop = df.select($"Hour", struct($"TotalValue", $"Category").alias("vs"))
  .groupBy($"hour")
  .agg(max("vs").alias("vs"))
  .select($"Hour", $"vs.Category", $"vs.TotalValue")

dfTop.show
// +----+--------+----------+
// |Hour|Category|TotalValue|
// +----+--------+----------+
// |   0|   cat26|      30.9|
// |   1|   cat67|      28.5|
// |   2|   cat56|      39.6|
// |   3|    cat8|      35.6|
// +----+--------+----------+

Das obige ist der detaillierte Inhalt vonWie wähle ich effizient die erste Zeile jeder Gruppe mit dem höchsten Wert in einem Spark-DataFrame aus?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!