Spark Dataset出现java.lang.UnsupportedOperationException，按Key取25行报错

咱们先拆解你代码里的问题，再给出针对性的修复方案：

1. 核心错误：分组键完全写错了 #

你原本的需求是按state（元组的第一个元素_._1）分组，但代码里写的是groupByKey(_._2)——这会把拼接后的整行字符串当成分组键，完全偏离了你的目标，这也是触发异常的核心原因。

2. 不必要的去重操作 #

你用了toSet，这会自动给组内记录去重，但你的需求是获取25行记录，而不是25个唯一的行，这不仅会丢失数据，还可能引发迭代器操作的异常。

3. 修复后的代码（贴合你的需求） #

如果只是想按state分组，每个组取25行原始记录，可以这样写：

import spark.implicits._

val record = file.map(rec => {
    // 用字符串插值让拼接更简洁易读
    val row = s"${rec.date},${rec.registrar},${rec.agency},${rec.state},${rec.district},${rec.subDistrict},${rec.pinCode},${rec.gender},${rec.age},${rec.aadharGenerated},${rec.rejected},${rec.mobileNo},${rec.email}"
    (rec.state, row)
})
.groupByKey(_._1) // 这里改成按state（元组第一个元素）分组
.mapGroups((state, rowsIter) => {
    // 直接从迭代器取前25条，不需要转Set去重
    (state, rowsIter.take(25).toList)
})
.collect()

record.foreach(println)

更高效的Window函数实现（大场景推荐） #

如果数据量较大，groupByKey可能会有shuffle性能瓶颈，用Window函数是更优的选择，还能支持按指定字段排序取Top N：

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

// 先转成DataFrame，方便使用Window函数
val df = file.toDF()

// 定义窗口：按state分区，可按需添加orderBy（比如按date取最新25条）
val windowSpec = Window.partitionBy("state").orderBy("date")

// 添加行号并筛选前25条
val top25PerState = df.withColumn("row_num", row_number().over(windowSpec))
                      .where("row_num <= 25")
                      .drop("row_num")

// 如需转成你之前的字符串格式，可再做拼接处理
top25PerState.map(row => {
    val state = row.getAs[String]("state")
    val rowStr = row.mkString(",")
    (state, rowStr)
}).collect().foreach(println)

关于异常的说明 #

你之前的groupByKey(_._2)会把每个不同的整行字符串当成独立分组，当分组数量极大时，Spark迭代器的某些操作（比如转Set）会因为内存限制或迭代器不支持的操作触发UnsupportedOperationException，修正分组键后这个问题就会消失。

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