如何使用Spark Dataset GroupBy()？Hive表取各id最新记录咨询

嗨，针对你这个要给每个id取updated_dt最大的完整记录的需求，我来给你详细说说两种可行的实现方式，包括你提到的RDD方案，再给你推荐个更适合大数据场景的优化方案～

按照你说的思路，先把Hive表数据转成对应case class的RDD，再分组筛选最大值，具体代码如下：

// 定义与Hive表结构对应的case class
case class UserRecord(id: BigInt, name: String, updated_dt: BigInt)

// 从Hive读取数据并转换为RDD[UserRecord]
val hiveDataRDD = spark.sql("SELECT id, name, updated_dt FROM your_hive_table").rdd
  .map(row => UserRecord(
    row.getAs[BigInt]("id"),
    row.getAs[String]("name"),
    row.getAs[BigInt]("updated_dt")
  ))

// 按id分组，然后提取每组中updated_dt最大的那条记录
val latestRecordsRDD = hiveDataRDD.groupBy(_.id)
  .mapValues(records => records.maxBy(_.updated_dt))
  .values

// 可选：将结果转回DataFrame并写入Hive表
latestRecordsRDD.toDF().write.mode("overwrite").saveAsTable("your_target_table")

这个方案逻辑很直观，但要注意：groupBy操作会触发shuffle，当数据量特别大的时候，性能可能会受影响。

在大数据场景下，用窗口函数的方式效率更高，而且代码更简洁，不需要手动转换RDD和case class，具体实现：

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

// 直接读取Hive表为DataFrame
val hiveDataDF = spark.sql("SELECT id, name, updated_dt FROM your_hive_table")

// 定义窗口规则：按id分区，按updated_dt降序排序
val idWindow = Window.partitionBy("id").orderBy(desc("updated_dt"))

// 给每个分区的记录添加行号，筛选行号为1的记录（即updated_dt最大的那条）
val latestRecordsDF = hiveDataDF
  .withColumn("row_rank", row_number().over(idWindow))
  .filter("row_rank = 1")
  .drop("row_rank")

// 写入结果到Hive
latestRecordsDF.write.mode("overwrite").saveAsTable("your_target_table")

额外说明： #

• 如果同一个id存在多条updated_dt相同的最大记录，方案一的maxBy会随机返回其中一条；方案二用row_number()也只会保留一条，要是想保留所有最大值记录，可以把row_number()换成rank()或dense_rank()。
• 注意数据类型匹配：Hive的bigint对应Spark中的Long或BigInt，代码里要对应正确，避免类型转换错误。

内容的提问来源于stack exchange，提问作者michelle