Spark报错：Dataset转换与操作仅能由Driver调用，求解决方案

报错信息 #

Dataset的转换与操作仅能由Driver调用，不可在其他转换内部执行；例如rdd1.map(x => rdd2.values.count() * x)，详见SPARK-28702

问题代码 #

do {
        val size = accumulatorList.value.size()
        val current = accumulatorList.value.get(size -1)
        joinoutput = current.join(broadvast.value, current.col("A") === broadvast.value.col("A"))
        .map {x =>
                
                    
        val gig = x._1.getAs("Name") + "|" + x._1.getAs("PIP")
            
        MyData(
        x._2.getAs("Name"),
        x._2.getAs("Place"),
        x._2.getAs("Phone"),
        x._2.getAs("Doc"),
        gig 
        )}          
        accumulatorList.add(joinoutput)
        joincount = joinoutput.count()

}

while (joincount >0 )

场景说明 #

报错出现在do块的join操作环节。代码在本地/集群环境运行正常，但构建测试用例时失败；测试用例在本地运行也正常，仅构建时出现问题。

问题根源 #

1. 误用Accumulator存储Dataset：Spark的Accumulator设计初衷是用于Driver端对分布式任务中的数据做累加统计（比如计数、求和），并不适合存储Dataset这类包含执行计划的复杂对象。当你把Dataset存入Accumulator后，循环中取出执行join操作时，构建环境的Spark执行计划解析器会误判该操作是在分布式转换内部触发的Driver端Dataset操作，触发SPARK-28702对应的检查规则。
2. 构建环境的严格校验机制：本地/集群运行时可能未开启严格的执行计划校验，而构建测试环境（如CI/CD流程）可能启用了更严格的Spark配置，导致原本隐藏的执行逻辑问题被暴露。

解决方法 #

1. 用普通变量替换Accumulator存储Dataset #

放弃使用Accumulator保存Dataset，改用普通列表或变量维护循环结果集，示例修改如下：

// 用普通List存储历史Dataset，替代Accumulator
var dsList = List(initialDataset) // initialDataset为循环起始的Dataset
var joincount = 1

do {
    val current = dsList.last
    val joinoutput = current.join(broadvast.value, current.col("A") === broadvast.value.col("A"))
        .map { x =>
            val gig = x._1.getAs("Name") + "|" + x._1.getAs("PIP")
            MyData(
                x._2.getAs("Name"),
                x._2.getAs("Place"),
                x._2.getAs("Phone"),
                x._2.getAs("Doc"),
                gig
            )
        }
    dsList = dsList :+ joinoutput
    joincount = joinoutput.count()
} while (joincount > 0)

2. 避免依赖Accumulator的分布式状态 #

Accumulator的value方法仅适合获取简单累加结果，存储复杂对象会引发序列化和执行计划解析问题，完全不适合这种迭代式的Dataset处理场景。

3. 对齐构建环境的Spark配置 #

检查构建测试环境中是否设置了严格校验类的Spark配置（如spark.sql.optimizer.excludeRules相关配置），若有可调整为与本地运行一致的配置，但核心解决思路仍是修正Accumulator的误用问题。

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