如何遍历Scala DataFrame行并将列值存入变量用于循环内操作

你当前的代码已经通过as[cls_Employee]将DataFrame转为强类型Dataset，每个遍历元素都是cls_Employee类的实例，只需在foreach的匿名函数内部完成变量赋值和函数调用即可，具体修改如下：

import spark.implicits._
import org.apache.spark.sql._

case class cls_Employee(name:String, sector:String, age:Int)
val df = Seq(
  cls_Employee("Andy","aaa", 20), 
  cls_Employee("Berta","bbb", 30), 
  cls_Employee("Joe","ccc", 40)
).toDF()

// 自定义示例函数，用于演示调用
def processEmployee(name: String, age: Int): Unit = {
  println(s"处理员工信息：姓名=$name，年龄=$age")
}

// 遍历Dataset，提取列值到变量并执行操作
df.as[cls_Employee].take(df.count.toInt).foreach(t => {
  // 将列值存入变量
  val empName = t.name
  val empSector = t.sector
  val empAge = t.age
  
  // 打印变量内容
  println(s"姓名：$empName，部门：$empSector，年龄：$empAge")
  
  // 调用自定义函数
  processEmployee(empName, empAge)
})

关键说明 #

• 因为as[cls_Employee]做了类型转换，遍历的每个t都是cls_Employee的实例，直接通过t.属性名就能获取对应列的值，赋值给变量和操作普通Scala类完全一致。
• 自定义函数可以直接接收这些变量作为参数调用，逻辑和普通Scala函数调用无差异。

注意事项 #

• take(df.count.toInt)会把全量数据拉到Driver节点内存中，仅适合小数据量场景。如果是大数据量生产环境，推荐使用Dataset的分布式foreach算子（无需take），它会在Executor节点并行处理数据，避免Driver内存溢出：

// 分布式遍历（适合大数据量）
df.as[cls_Employee].foreach(t => {
  val empName = t.name
  val empAge = t.age
  processEmployee(empName, empAge)
})

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