Как выбрать первый ряд каждой группы?

143

У меня есть DataFrame, сгенерированный следующим образом:

df.groupBy($"Hour", $"Category")
  .agg(sum($"value") as "TotalValue")
  .sort($"Hour".asc, $"TotalValue".desc))

Результаты выглядят так:

+----+--------+----------+
|Hour|Category|TotalValue|
+----+--------+----------+
|   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|
| ...|    ....|      ....|
+----+--------+----------+

Как видите, DataFrame упорядочен по Hourвозрастанию, а затем по TotalValueубыванию.

Я хотел бы выбрать верхний ряд каждой группы, т.е.

из группы Час == 0 выберите (0, cat26,30.9)
из группы Час == 1 выберите (1, cat67,28.5)
из группы Час == 2 выберите (2, cat56,39.6)
и так далее

Таким образом, желаемый результат будет:

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

Может быть удобно иметь возможность выбирать N верхних строк каждой группы.

Любая помощь высоко ценится.

sql scala apache-spark dataframe apache-spark-sql Рами
источник

232

Оконные функции :

Нечто подобное должно сработать:

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|
// +----+--------+----------+

Этот метод будет неэффективным в случае значительного перекоса данных.

Агрегирование простого SQL с последующимjoin :

В качестве альтернативы вы можете объединиться с агрегированным фреймом данных:

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|
// +----+--------+----------+

В нем будут сохранены повторяющиеся значения (если существует более одной категории в час с одинаковым общим значением). Вы можете удалить их следующим образом:

dfTopByJoin
  .groupBy($"hour")
  .agg(
    first("category").alias("category"),
    first("TotalValue").alias("TotalValue"))

Использование заказа черезstructs :

Аккуратный, хотя и не очень хорошо протестированный, трюк, который не требует объединений или оконных функций:

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|
// +----+--------+----------+

С помощью DataSet API (Spark 1.6+, 2.0+):

Спарк 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 или более поздняя версия :

df.as[Record]
  .groupByKey(_.Hour)
  .reduceGroups((x, y) => if (x.TotalValue > y.TotalValue) x else y)

Последние два метода могут использовать объединение на стороне карты и не требуют полного перемешивания, поэтому большую часть времени должны демонстрировать лучшую производительность по сравнению с оконными функциями и объединениями. Они также могут использоваться со структурированной потоковой передачей в completedрежиме вывода.

Не используйте :

df.orderBy(...).groupBy(...).agg(first(...), ...)

Это может показаться на работу (особенно в localрежиме) , но это ненадежно (см СПАРК-16207 , кредиты Tzach Зоар для увязки соответствующей проблемы JIRA и SPARK-30335 ).

Это же примечание относится к

df.orderBy(...).dropDuplicates(...)

который внутренне использует эквивалентный план выполнения.

zero323
источник

3

Похоже, что после spark 1.6 это row_number () вместо rowNumber

Адам Шалуча

О Не используйте df.orderBy (...). GropBy (...). При каких обстоятельствах мы можем положиться на orderBy (...)? или если мы не можем быть уверены, что orderBy () даст правильный результат, какие у нас есть альтернативы?

Игнасио Алорре

Я мог бы что-то пропустить, но в целом рекомендуется избегать groupByKey , вместо этого следует использовать reduByKey. Кроме того, вы будете сохранять одну строку.

Томас

3

@Thomas, избегающий groupBy / groupByKey, - это просто когда имеешь дело с RDD, вы заметите, что в API набора данных даже нет функции reduByKey.

Soote

@Thomas DataFrame / Группа данных Поведение / оптимизация

Alper t. Туркер

16

Для Spark 2.0.2 с группировкой по нескольким столбцам:

import org.apache.spark.sql.functions.row_number
import org.apache.spark.sql.expressions.Window

val w = Window.partitionBy($"col1", $"col2", $"col3").orderBy($"timestamp".desc)

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

Антонин Хосковец
источник

8

Это точно такой же ответ ноль323 , но в виде SQL-запроса.

Предполагая, что датафрейм создан и зарегистрирован как

df.createOrReplaceTempView("table")
//+----+--------+----------+
//|Hour|Category|TotalValue|
//+----+--------+----------+
//|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      |
//+----+--------+----------+

Оконная функция:

sqlContext.sql("select Hour, Category, TotalValue from (select *, row_number() OVER (PARTITION BY Hour ORDER BY TotalValue DESC) as rn  FROM table) tmp where rn = 1").show(false)
//+----+--------+----------+
//|Hour|Category|TotalValue|
//+----+--------+----------+
//|1   |cat67   |28.5      |
//|3   |cat8    |35.6      |
//|2   |cat56   |39.6      |
//|0   |cat26   |30.9      |
//+----+--------+----------+

Агрегирование простого SQL с последующим соединением:

sqlContext.sql("select Hour, first(Category) as Category, first(TotalValue) as TotalValue from " +
  "(select Hour, Category, TotalValue from table tmp1 " +
  "join " +
  "(select Hour as max_hour, max(TotalValue) as max_value from table group by Hour) tmp2 " +
  "on " +
  "tmp1.Hour = tmp2.max_hour and tmp1.TotalValue = tmp2.max_value) tmp3 " +
  "group by tmp3.Hour")
  .show(false)
//+----+--------+----------+
//|Hour|Category|TotalValue|
//+----+--------+----------+
//|1   |cat67   |28.5      |
//|3   |cat8    |35.6      |
//|2   |cat56   |39.6      |
//|0   |cat26   |30.9      |
//+----+--------+----------+

Используя упорядочение по структурам:

sqlContext.sql("select Hour, vs.Category, vs.TotalValue from (select Hour, max(struct(TotalValue, Category)) as vs from table group by Hour)").show(false)
//+----+--------+----------+
//|Hour|Category|TotalValue|
//+----+--------+----------+
//|1   |cat67   |28.5      |
//|3   |cat8    |35.6      |
//|2   |cat56   |39.6      |
//|0   |cat26   |30.9      |
//+----+--------+----------+

Путь DataSets и не делать s такие же, как в оригинальном ответе

Рамеш Махарджан
источник

2

Шаблон сгруппирован по ключам => сделать что-то для каждой группы, например, уменьшить => вернуться к кадру данных

Я думал, что абстракция Dataframe немного громоздка в этом случае, поэтому я использовал функциональность RDD

 val rdd: RDD[Row] = originalDf
  .rdd
  .groupBy(row => row.getAs[String]("grouping_row"))
  .map(iterableTuple => {
    iterableTuple._2.reduce(reduceFunction)
  })

val productDf = sqlContext.createDataFrame(rdd, originalDf.schema)

Резиновая утка
источник

1

Решение ниже делает только один groupBy и извлекает строки вашего фрейма данных, которые содержат maxValue в одном кадре. Нет необходимости в дальнейшем присоединении или Windows.

import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.DataFrame

//df is the dataframe with Day, Category, TotalValue

implicit val dfEnc = RowEncoder(df.schema)

val res: DataFrame = df.groupByKey{(r) => r.getInt(0)}.mapGroups[Row]{(day: Int, rows: Iterator[Row]) => i.maxBy{(r) => r.getDouble(2)}}

elghoto
источник

Но это сначала все перемешивает. Это вряд ли улучшение (может быть, не хуже, чем оконные функции, в зависимости от данных).

Альпер т. Туркер

у вас есть первое место в группе, что вызовет случайное перемешивание. Это не хуже, чем оконная функция, потому что в оконной функции она будет оценивать окно для каждой отдельной строки в кадре данных.

elghoto

1

Хороший способ сделать это с помощью API dataframe - использовать логику argmax следующим образом

  val df = 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")

  df.groupBy($"Hour")
    .agg(max(struct($"TotalValue", $"Category")).as("argmax"))
    .select($"Hour", $"argmax.*").show

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

randal25
источник

0

Здесь вы можете сделать это -

   val data = df.groupBy("Hour").agg(first("Hour").as("_1"),first("Category").as("Category"),first("TotalValue").as("TotalValue")).drop("Hour")

data.withColumnRenamed("_1","Hour").show

Шубхам Агравал
источник

-2

Мы можем использовать оконную функцию rank () (где вы выбрали бы rank = 1) rank просто добавляет число для каждой строки группы (в данном случае это будет час)

вот пример. (из https://github.com/jaceklaskowski/mastering-apache-spark-book/blob/master/spark-sql-functions.adoc#rank )

val dataset = spark.range(9).withColumn("bucket", 'id % 3)

import org.apache.spark.sql.expressions.Window
val byBucket = Window.partitionBy('bucket).orderBy('id)

scala> dataset.withColumn("rank", rank over byBucket).show
+---+------+----+
| id|bucket|rank|
+---+------+----+
|  0|     0|   1|
|  3|     0|   2|
|  6|     0|   3|
|  1|     1|   1|
|  4|     1|   2|
|  7|     1|   3|
|  2|     2|   1|
|  5|     2|   2|
|  8|     2|   3|
+---+------+----+

Василе Сурду
источник

Как выбрать первый ряд каждой группы?

Ответы: