Apache Spark Scala遍历DataFrame行生成新DataFrame时MutableList为空问题

这是Spark新手很容易踩的分布式架构认知坑，核心原因是你误解了Spark中foreach操作的执行逻辑，以及Driver和Executor之间的数据隔离机制。

为什么tList最后是空的？ #

Spark是分布式计算框架，你的代码会分成两部分执行：

• Driver端：初始化mutable.MutableList[IdCount]()的逻辑，以及最后调用tList.toDF().show()的逻辑，都是在提交任务的Driver节点上执行的。
• Executor端：df.foreach(row => {...})里的循环逻辑，是被分发到集群的各个Executor节点上并行执行的。每个Executor的Task都会拿到df的一部分数据，并且会创建自己的tList副本（因为Driver的变量会被序列化传到Executor）。你在循环里打印的record和tList，其实是每个Executor本地的副本，这些修改根本不会同步回Driver端的原始tList。所以当Driver执行到tList.toDF().show()时，原始的tList还是空的。

排查验证方法 #

你可以在foreach里加一行打印当前节点的代码，直观看到执行位置的差异：

df.foreach(row => {
    // ... 你的原有代码
    println(s"当前执行节点: ${java.net.InetAddress.getLocalHost.getHostName}")
})

运行后你会发现，循环里的打印信息来自不同的节点，而Driver端的tList完全没被修改。

正确的解决方案 #

Spark的设计理念是尽量用分布式转换操作处理数据，而不是把数据拉到Driver端处理。这里有几种更合理的实现方式：

方案1：用map转换生成新DataFrame（推荐）

这是最符合Spark范式的做法，直接在分布式环境下转换每行数据，不需要依赖Driver端的可变集合：

import org.apache.spark.sql.DataFrame
case class IdCount(mailid: String, wordCount: Int)

def analyseDF(df: DataFrame): DataFrame = {
    df.map { row =>
        val mailid = row.getString(0)
        // 注意：如果body列有null，要先判断避免空指针
        val wordCnt = Option(row.getString(5)).map(_.split("\\s+").size).getOrElse(0)
        IdCount(mailid, wordCnt)
    }.toDF()
}

// 调用示例
analyseDF(yourOriginalDF).show()

方案2：用Spark SQL内置函数实现（更高效）

不需要自定义case class，直接用Spark的内置函数完成计算，性能更好：

import org.apache.spark.sql.functions.{col, size, split}

df.select(
    col("id").alias("mailid"),
    // 用\\s+匹配任意空白字符，避免多个空格导致统计错误
    size(split(col("body"), "\\s+")).alias("wordCount")
).show()

方案3：如果必须拉取到Driver端（仅适用于小数据量）

如果你的数据量很小，可以先把所有行收集到Driver，再处理：

def analyseDF(df: DataFrame): Unit = {
    val tList = df.collect().map { row =>
        val mailid = row.getString(0)
        val wordCnt = row.getString(5).split("\\s+").size
        IdCount(mailid, wordCnt)
    }.toList
    tList.toDF().show()
}

⚠️ 注意：collect()会把整个DataFrame的数据拉到Driver内存，数据量大时会直接OOM，所以只适合小数据集。

总结 #

• 永远不要在Spark的分布式操作（如foreach、mapPartitions）里修改Driver端的可变集合，这是无效且容易出错的。
• 优先使用Spark的分布式转换操作（map、select等）处理数据，既保证正确性又能利用集群的并行计算能力。

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