如何选择 Spark DataFrame 中每个组的第一行？

选择每个分组的第一行

为了根据特定的排序标准检索每个分组的第一行，您可以使用几种方法：

窗口函数

import org.apache.spark.sql.functions.{row_number, max, broadcast}
import org.apache.spark.sql.expressions.Window

// 创建一个用于分区和排序的窗口对象
val w = Window.partitionBy($"Hour").orderBy($"TotalValue".desc)

// 添加一个排名列来标识每个分组的第一行
val dfTop = df.withColumn("rn", row_number.over(w))

// 过滤排名为1的行
dfTop.where($"rn" === 1).drop("rn")

普通SQL聚合和连接

// 聚合以查找每个小时的最大值
val dfMax = df.groupBy($"Hour".as("max_hour")).agg(max($"TotalValue").as("max_value"))

// 将原始DataFrame与聚合后的DataFrame连接
val dfTopByJoin = df.join(broadcast(dfMax), ($"Hour" === $"max_hour") && ($"TotalValue" === $"max_value"))

// 删除不必要的列
dfTopByJoin.drop("max_hour").drop("max_value")

结构体排序

// 为包含TotalValue和Category的结构体定义别名
val vs = struct($"TotalValue", $"Category").alias("vs")

// 按Hour分组并查找每个分组的最大结构体
val dfTop = df.select($"Hour", vs).groupBy($"Hour").agg(max(vs).alias("vs"))

// 从最大结构体中提取Category和TotalValue
dfTop.select($"Hour", $"vs.Category", $"vs.TotalValue")

使用DataFrame API

// 为DataFrame定义一个自定义类
case class Record(Hour: Integer, Category: String, TotalValue: Double)

// 将DataFrame转换为自定义类
val dfRecords = df.as[Record]

// 按Hour分组并减少以查找TotalValue最大的记录
val dfTopRecords = dfRecords.groupByKey(_.Hour).reduce((x, y) => if (x.TotalValue > y.TotalValue) x else y)

// 转换回DataFrame
dfTopRecords.toDF

以上是如何选择 Spark DataFrame 中每个组的第一行？的详细内容。更多信息请关注PHP中文网其他相关文章！