如何选择每组的第一行？

Question

如何选择每组的第一行？

我生成了一个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按降序排列。

我想选择每一组的顶部行，即

• 从组h=0选择(0，cat26，30.9)
• 从小时组=1选择(1，cat67，28.5)
• 从小时组=2选择(2，cat 56，39.6)
• 诸若此类

因此，期望的输出是：

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

也可以方便地选择每个组的前N行。

任何帮助都是非常感谢的。

Answer 1

对于星火2.0.2，按多列分组：import org.apache.spark.sql.functions.row_numberimport org.apache.spark.sql.expressions.Windowval w = Window.partitionBy($"col1", $"col2", $"col3").orderBy($"timestamp".desc)val refined_df = df.withColumn("rn", row_number.over(w)).where($"rn" === 1).drop("rn")

Answer 2

这和零323氏回答而是以SQL查询的方式。假设创建了dataframe并将其注册为假设创建了dataframe并将其注册为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      |//+----+--------+----------+数据集方法和别这样S与原答案相同。