Spark类型安全agg函数报错：GenericRowWithSchema无法转为JosmSalesRecord

问题根源 #

你遇到的ClassCastException是因为typed.sum系列类型安全聚合函数要求操作强类型的Dataset[JosmSalesRecord]，而当前的aggregateSum是普通DataFrame（即Dataset[Row]），其元素是GenericRowWithSchema，无法直接转换为JosmSalesRecord实例，进而触发类型转换错误。

解决步骤 #

1. 将DataFrame转换为强类型Dataset[JosmSalesRecord] #

确保你的数据集是基于样例类的强类型Dataset，而非Row类型的DataFrame。如果是从外部数据源读取的DataFrame，需先完成类型转换，同时要保证字段名与样例类匹配：

// 假设aggregateSum是原始DataFrame
// 若原始字段是下划线命名，先映射为样例类的驼峰字段名
val renamedDF = aggregateSum
  .withColumnRenamed("order_id", "orderId")
  .withColumnRenamed("total_order_sales", "totalOrderSales")
  .withColumnRenamed("total_order_count", "totalOrderCount")

// 转换为强类型Dataset
val aggregateSumDS: Dataset[JosmSalesRecord] = renamedDF.as[JosmSalesRecord]

2. 使用typed聚合函数执行聚合 #

基于强类型Dataset调用typed.sum系列函数，这些函数返回TypedColumn，直接聚合后即可得到正确结果：

import org.apache.spark.sql.expressions.scalalang.typed._

val pv = aggregateSumDS.agg(
  sum[JosmSalesRecord](_.totalOrderSales).alias("totalOrderSales"),
  sumLong[JosmSalesRecord](_.totalOrderCount).alias("totalOrderCount")
)

pv.show()

关键说明 #

typed系列聚合函数的核心价值就是类型安全，其设计目标是直接操作具体的样例类实例，而非通用的Row对象。如果直接在DataFrame上使用这类函数，Spark会尝试强制将Row转换为样例类，这必然会触发类型转换异常。

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