Рассчитать максимальное количество наблюдений на группу

Я использую Spark 1.6.2.

Мне нужно найти максимальное количество для каждой группы.

val myData = Seq(("aa1", "GROUP_A", "10"),("aa1","GROUP_A", "12"),("aa2","GROUP_A", "12"),("aa3", "GROUP_B", "14"),("aa3","GROUP_B", "11"),("aa3","GROUP_B","12" ),("aa2", "GROUP_B", "12"))

val df = sc.parallelize(myData).toDF("id","type","activity")

Давайте сначала подсчитаем количество наблюдений на группу:

df.groupBy("type","id").count.show

+-------+---+-----+
|   type| id|count|
+-------+---+-----+
|GROUP_A|aa1|    2|
|GROUP_A|aa2|    1|
|GROUP_B|aa2|    1|
|GROUP_B|aa3|    3|
+-------+---+-----+

Это ожидаемый результат:

+--------+----+-----+
|type    |  id|count|
+----+--------+-----+
| GROUP_A| aa1|    2|
| GROUP_B| aa3|    3|
+--------+----+-----+

Я пробовал это, но это не работает:

df.groupBy("type","id").count.filter("count = 'max'").show

apache-spark scala apache-spark-1.6
person Dinosaurius    schedule 11.05.2017    source источник
comment
ваш ожидаемый результат не коррелирует с тем, что вы, кажется, описываете: если вы хотите максимальное количество для каждой группы, вы должны получить 12 для группы A и 14 для группы B, не так ли? И какой id вы ожидаете увидеть - тот, который соответствует записи с максимальным count? Просьба уточнить.   -  person Tzach Zohar    schedule 11.05.2017
comment
@TzachZohar: Нет, я не ищу максимальное значение. Я хочу сначала подсчитать количество наблюдений на группу, а затем выбрать максимум count на группу. Пожалуйста, смотрите мою обновленную тему. Я объяснил это шаг за шагом. Чтобы избежать путаницы, я переименовал начальные столбцы в .toDF("id","type","activity").   -  person Dinosaurius    schedule 11.05.2017

Ответы (3)


arrow_upward
2
arrow_downward

Чтобы получить «строку с максимальным значением столбца X» (а не только это максимальное значение), вы можете использовать этот небольшой трюк «сгруппировать» соответствующие столбцы вместе в struct, который содержит столбец порядка в качестве первого столбца, а затем вычисление max для этой структуры. Поскольку порядок struct «доминирует» над порядком его первого столбца, мы получим желаемый результат:

df.groupBy("id","type").count()                // get count per id and type
  .groupBy("type")                             // now group by type only
  .agg(max(struct("count", "id")) as "struct") // get maximum of (count, id) structs - since count is first, and id is unique - count will decide the ordering
  .select($"type", $"struct.id" as "id", $"struct.count" as "count") // "unwrap" structs
  .show()

// +-------+---+-----+
// |   type| id|count|
// +-------+---+-----+
// |GROUP_A|aa1|    2|
// |GROUP_B|aa3|    3|
// +-------+---+-----+
person Tzach Zohar    schedule 11.05.2017

arrow_upward
1
arrow_downward

Вы можете использовать функцию max после группы.

val myData = Seq(("aa1", "GROUP_A", "10"),("aa1","GROUP_A", "12"),("aa2","GROUP_A", "12"),("aa3", "GROUP_B", "14"),("aa3","GROUP_B", "11"),("aa3","GROUP_B","12" ),("aa2", "GROUP_B", "12"))

val df = sc.parallelize(myData).toDF("id","type","activity")

// count после groupby, а затем псевдоним для полей count, после чего найдите максимальное значение в поле cnt.

val newDF = df1.groupBy("type", "id").agg(count("*").alias("cnt"))

val df1 = newDF.groupBy("type").max("cnt").show

Теперь вы можете присоединиться к этим двум кадрам данных, чтобы получить результат.

df1.join(newDF.as("newDF"), col("cnt") === col("max(cnt)")).select($"newDF.*").show
person koiralo    schedule 11.05.2017
comment
Я обновил ответ. Раньше я ошибался из-за поля count в вашем фрейме данных. - person koiralo; 11.05.2017
comment
Спасибо. Столбец id не отображается в newDF. - person Dinosaurius; 11.05.2017
comment
Вы можете объединить два кадра данных с совпадающим количеством и получить результат. - person koiralo; 11.05.2017

arrow_upward
1
arrow_downward

Вы можете использовать функцию Window, чтобы найти max и удалить duplicates, объединив ответ @Tzach выше

val windowSpec = Window.partitionBy(col("type"))
import org.apache.spark.sql.functions._
df.groupBy("type","id").count()
  .withColumn("count", max(struct("count", "id")).over(windowSpec))
  .dropDuplicates("type")
  .select($"type", $"count.id" as "id", $"count.count" as "count").show

Спасибо

person Ramesh Maharjan    schedule 11.05.2017
comment
Не могли бы вы проверить мою обновленную тему, где я привел пример и ожидаемый результат. - person Dinosaurius; 11.05.2017
comment
@Dinosaurius, вы можете использовать window function, как указано выше. - person Ramesh Maharjan; 11.05.2017