Spark groupByKey后排序调用Aggregator聚合报类转换异常解决方案

报错原因说明 #

• 调用mapGroups后返回的是普通Dataset类型，不再是KeyValueGroupedDataset，直接调用原有的agg方法时，Spark不会将元组里的Seq/Iterator识别为分组待聚合的元素集合，会触发GenericRow强转case class的ClassCastException。
• Iterator是不可序列化的流式遍历接口，Spark没有提供对应的Encoder，无法作为Dataset的元素类型跨节点传输，因此返回Iterator的写法会直接报编码器缺失错误。
• 全局调用sortBy("ts")后再分组无法保证顺序，是因为全局排序会按ts做shuffle分发，相同主键的记录会被拆分到不同分区，聚合时自然无法保留排序结果。

方案1：分区预排序（无需修改现有Aggregator，性能最优） #

这个方案不需要改动你已经写好的自定义Aggregator逻辑，通过重分区+分区内排序保证同组数据顺序，是生产环境优先选择的写法：

// 先按主键重分区，保证同主键数据落在同一分区，再在分区内按主键+ts排序
val sortedInput = seq_users
  .repartition($"pk")
  .sortWithinPartitions($"pk", $"ts")
  .as[Player]

val grouped = sortedInput.groupByKey(_.pk)
val result = grouped.agg(BatchDedupe.toColumn.name("deduped"))
result.show(false)

执行后返回结果符合按ts顺序聚合的预期：

+--------------+--------------------------------+
|key           |deduped                         |
+--------------+--------------------------------+
|12121212121212|{10000004, null, 1985-01-02}    |
+--------------+--------------------------------+

原理说明：repartition($"pk")会将相同主键的所有记录shuffle到同一个executor分区，后续sortWithinPartitions仅在单分区内做排序，不会触发跨分区数据打乱，后续groupByKey聚合时，遍历同组记录的顺序和分区内排序顺序完全一致，Spark不会在该阶段破坏同分区内的记录顺序。

方案2：分组内手动排序+聚合（适合有复杂组内预处理的场景） #

如果需要在分组后做除排序外的其他自定义预处理，不需要强行将mapGroups的输出转回KeyValueGroupedDataset，直接在mapGroups内完成排序+聚合全流程即可：

val result = seq_users
  .groupByKey(_.pk)
  .mapGroups { case (pk, iter) =>
    // 组内按ts升序排序，如需取最新值可调用reverse改为降序
    val sortedIter = iter.toSeq.sortBy(_.ts).iterator
    // 手动执行自定义Aggregator的处理逻辑
    val buffer = BatchDedupe.zero
    while (sortedIter.hasNext) {
      BatchDedupe.reduce(buffer, sortedIter.next())
    }
    (pk, BatchDedupe.finish(buffer))
  }
  .toDF("key", "deduped")

result.show(false)

注意：该写法会将单个分组的所有记录加载到内存，如果单个主键对应的数据量极大（比如单key超过百万条记录），可能触发Executor OOM，这种场景优先选择方案1。

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