Funkcje okna :
Coś takiego powinno załatwić sprawę:
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|
// +----+--------+----------+
Ta metoda będzie nieefektywna w przypadku znacznego odchylenia danych.
Zwykła agregacja SQL, po której następujejoin
:
Alternatywnie możesz dołączyć do zagregowanej ramki danych:
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|
// +----+--------+----------+
Zachowa zduplikowane wartości (jeśli na godzinę jest więcej niż jedna kategoria o tej samej łącznej wartości). Możesz je usunąć w następujący sposób:
dfTopByJoin
.groupBy($"hour")
.agg(
first("category").alias("category"),
first("TotalValue").alias("TotalValue"))
Korzystanie z zamawiania powyżejstructs
:
Zgrabna, choć niezbyt dobrze przetestowana sztuczka, która nie wymaga łączenia ani funkcji okna:
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|
// +----+--------+----------+
Z DataSet API (Spark 1.6+, 2.0+):
Spark 1.6 :
case class Record(Hour: Integer, Category: String, TotalValue: Double)
df.as[Record]
.groupBy($"hour")
.reduce((x, y) => if (x.TotalValue > y.TotalValue) x else y)
.show
// +---+--------------+
// | _1| _2|
// +---+--------------+
// |[0]|[0,cat26,30.9]|
// |[1]|[1,cat67,28.5]|
// |[2]|[2,cat56,39.6]|
// |[3]| [3,cat8,35.6]|
// +---+--------------+
Spark 2.0 lub nowszy :
df.as[Record]
.groupByKey(_.Hour)
.reduceGroups((x, y) => if (x.TotalValue > y.TotalValue) x else y)
Dwie ostatnie metody mogą wykorzystywać łączenie strony mapy i nie wymagają pełnego tasowania, więc przez większość czasu powinny wykazywać lepszą wydajność w porównaniu z funkcjami okna i złączeniami. Te laski mogą być również używane z Structured Streaming w completed
trybie wyjściowym.
Nie używaj :
df.orderBy(...).groupBy(...).agg(first(...), ...)
Może się wydawać, że działa (szczególnie w local
trybie), ale jest zawodne (zobacz SPARK-16207 , podziękowania dla Tzacha Zohara za powiązanie odpowiedniego problemu z JIRA i SPARK-30335 ).
Ta sama uwaga dotyczy
df.orderBy(...).dropDuplicates(...)
który wewnętrznie używa równoważnego planu wykonania.
W przypadku Spark 2.0.2 z grupowaniem według wielu kolumn:
źródło
To jest dokładnie to samo z zero323 „s odpowiedzi , ale w SQL kwerendy sposób.
Zakładając, że dataframe jest tworzona i rejestrowana jako
Funkcja okna:
Zwykła agregacja SQL, po której następuje łączenie:
Korzystanie z porządkowania na strukturach:
DataSets sposób i nie robią są takie same jak w oryginalnej odpowiedzi
źródło
Wzorzec jest grupowany według klawiszy => zrób coś dla każdej grupy, np. Zredukuj => wróć do ramki danych
Myślałem, że abstrakcja Dataframe jest w tym przypadku trochę kłopotliwa, więc użyłem funkcjonalności RDD
źródło
Poniższe rozwiązanie wykonuje tylko jedną grupę GroupBy i wyodrębnia wiersze ramki danych, które zawierają wartość maxValue w jednym ujęciu. Nie ma potrzeby dalszych połączeń ani systemu Windows.
źródło
Dobrym sposobem na zrobienie tego z interfejsem dataframe API jest użycie takiej logiki argmax
źródło
Tutaj możesz to zrobić -
źródło
Możemy użyć funkcji okna rank () (gdzie wybrałbyś rank = 1) rank po prostu dodaje liczbę dla każdego wiersza grupy (w tym przypadku byłaby to godzina)
oto przykład. (z https://github.com/jaceklaskowski/mastering-apache-spark-book/blob/master/spark-sql-functions.adoc#rank )
źródło