Spark中向UDF传递Array[Item]报错：Schema for type Dataset[Row] is not supported

你遇到的Schema for type org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] is not supported异常，核心原因是Spark UDF不支持返回DataFrame/Dataset这类分布式数据集类型。UDF是针对单条记录的本地处理逻辑，只能返回Spark可序列化的本地数据类型（如基本类型、数组、结构体、Map等）。

以下是针对你的场景的几种可行解决方案：

方案1：直接在UDF中返回过滤后的数组 #

如果你的需求只是对每个id对应的rowList做过滤（比如筛选符合条件的Item），无需在UDF中创建DataFrame，直接操作数组即可：

// 定义过滤逻辑的UDF，返回Array[Item]
val filterItemsUdf = udf((items: Array[Item]) => {
  // 示例：过滤counter大于5的Item
  items.filter(_.counter > 5)
})

// 应用UDF生成过滤后的列
val resultDf = curatedDf.withColumn("filteredList", filterItemsUdf(col("rowList")))
resultDf.show(false)

如果需要将过滤结果转为Row结构，也可以直接在UDF中转换：

val filterToRowUdf = udf((items: Array[Item]) => {
  items.filter(_.counter > 5)
       .map(item => Row(item.id, item.subgroup, item.pid, item.counter))
})

方案2：返回统计结果（而非完整数据集） #

如果你的内部逻辑需要做统计（比如每个id下的counter总和、pid出现次数），可以在UDF中直接计算并返回结构体类型：

// 定义统计结果的样例类
case class Stats(totalCounter: Long, pidCount: Map[String, Int]) extends Serializable

// 实现统计逻辑的UDF
val statsUdf = udf((items: Array[Item]) => {
  val total = items.map(_.counter).sum
  val pidCounts = items.groupBy(_.pid).mapValues(_.length)
  Stats(total, pidCounts)
})

// 生成统计列
val statsDf = curatedDf.withColumn("stats", statsUdf(col("rowList")))
statsDf.show(false)

方案3：避免collect_list，使用Spark原生操作（更高效） #

collect_list会将每个分组的所有数据拉到Executor内存中，数据量大时容易触发OOM。优先使用Spark原生的聚合或窗口函数处理，性能更优：

示例1：按id分组去重+过滤 #

如果需要过滤重复的(pid, subgroup)记录，用窗口函数标记后过滤：

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

val windowSpec = Window.partitionBy("id", "subgroup", "pid").orderBy("counter")

val filteredDf = df
  .withColumn("rn", row_number().over(windowSpec))
  .filter("rn = 1") // 保留每组第一条记录
  .drop("rn")

示例2：按id直接聚合统计 #

如果需要汇总每个id的统计信息，直接用groupBy+聚合函数：

import org.apache.spark.sql.functions.{sum, countDistinct}

val aggDf = df
  .groupBy("id")
  .agg(
    sum("counter").as("total_counter"),
    countDistinct("pid").as("distinct_pid_count")
  )

关键注意事项 #

1. UDF仅支持返回本地数据类型，禁止返回DataFrame/Dataset这类分布式对象。
2. 尽量避免collect_list，尤其是大数据集场景，优先使用Spark原生算子处理。
3. 若必须处理分组后的数组，需确保数组大小在Executor内存承受范围内，必要时调整spark.sql.shuffle.partitions和Executor内存配置。

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