Scala基于key.csv列键拆分大CSV为多文件报错求助

错误原因分析 #

你触发的SparkException核心问题是：在Spark分布式Dataset的转换操作（如foreach）内部调用了另一个Dataset的IO/转换操作。Spark的executor节点无法执行驱动端专属的SparkContext/Dataset操作，所有涉及Spark核心API的操作必须由驱动端发起。

典型错误代码示例 #

// 错误写法：在分布式foreach中执行Spark写入操作
val keyDF = spark.read.csv("key.csv").select("model_name")
val inputDF = spark.read.csv("Input.csv")

keyDF.foreach(row => {
  val modelName = row.getAs[String]("model_name")
  // 此处代码会在executor端执行，触发SparkException
  inputDF.filter(s"model_name = '$modelName'")
    .select("ID", s"${modelName}_col1", s"${modelName}_col2")
    .write.csv(s"$modelName.csv")
})

对应报错信息 #

SparkException: Dataset transformations and actions can only be invoked by the driver, not inside of other Dataset transformations;

正确实现方案 #

核心思路是：先将需要拆分的model_name列表拉取到驱动端本地，再在驱动端循环执行每个模型的数据过滤与写入操作。

完整可运行代码 #

import org.apache.spark.sql.SparkSession

object SplitCsvByModel {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("SplitCsvByModel")
      .master("local[*]") // 本地测试用，生产环境请移除
      .getOrCreate()

    import spark.implicits._

    // 读取key.csv，将model_name拉取到驱动端本地集合
    val modelNames = spark.read
      .option("header", "true") // 假设csv文件包含表头
      .csv("key.csv")
      .select("model_name")
      .as[String]
      .collect() // 关键：将分布式数据转为驱动端本地数组

    // 读取Input.csv源数据
    val inputDF = spark.read
      .option("header", "true")
      .csv("Input.csv")

    // 驱动端循环处理每个模型
    modelNames.foreach { modelName =>
      // 筛选当前模型的相关列（示例按前缀匹配，可根据实际列名规则调整）
      val targetColumns = Seq("ID") ++ inputDF.columns.filter(_.startsWith(modelName))
      
      inputDF
        .filter(s"model_name = '$modelName'") // 过滤当前模型的数据
        .select(targetColumns.head, targetColumns.tail: _*)
        .write
        .option("header", "true")
        .mode("overwrite") // 可选：覆盖已存在的文件
        .csv(s"./${modelName}.csv")
    }

    spark.stop()
  }
}

关键要点说明 #

1. collect()的作用：将分布式存储的model_name数据拉取到驱动端本地集合，后续的foreach是普通Scala循环，而非Spark分布式操作，避免了executor调用驱动端API的问题。
2. 列匹配逻辑：示例中用startsWith(modelName)匹配模型相关列，可根据实际业务调整（如固定后缀、特定命名规则等）。
3. 性能注意事项：如果model_name数量极大，collect()可能占用驱动端较多内存，此时可考虑分批处理或优化过滤逻辑。

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