如何选择每组的第一行？

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_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")

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与原答案相同。