Spark Scala遍历DataFrame行提取CSV值时控制台无输出问题排查

• CSV读取未启用表头解析：当前代码直接用spark.read.csv读取文件，默认不会将CSV第一行作为列名，DataFrame的列名会被设为_c0、_c1这类默认值。此时用row.getAs[String]("number")找不到对应列，代码异常因分布式执行特性，不会直接在Driver控制台显示。
• 分布式执行的输出特性：df.foreach是分布式Action操作，其中的println代码会在Worker节点执行，输出会打印到Worker的日志文件中，而非Driver端控制台，所以本地看不到输出。

针对上述问题，给出两种可行修改方案：

方案一：正确读取CSV并将输出拉到Driver端（小数据量适用） #

如果CSV数据量不大，可先将DataFrame内容收集到Driver节点再遍历，确保println内容显示在控制台：

import org.apache.spark.sql.SparkSession

object ExampleDataFrameIteration {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("ExampleDataFrameIteration").getOrCreate()

    // 启用表头解析，将CSV第一行作为列名
    val df = spark.read.option("header", "true").csv("path/to/file.csv")

    // 将数据拉到Driver节点后遍历（仅适用于小数据量）
    df.collect().foreach { row =>
      val number = row.getAs[String]("number")
      val account = row.getAs[String]("account")
      println(s"number: $number, Account: $account")
    }

    spark.stop()
  }
}

方案二：分布式执行时输出到Worker日志（大数据量适用） #

如果数据量较大，不适合拉到Driver端，可保持分布式遍历，通过Spark日志系统查看输出：

import org.apache.spark.sql.SparkSession
import org.apache.log4j.Logger

object ExampleDataFrameIteration {
  private val logger = Logger.getLogger(this.getClass)

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("ExampleDataFrameIteration").getOrCreate()

    val df = spark.read.option("header", "true").csv("path/to/file.csv")

    df.foreach { row =>
      val number = row.getAs[String]("number")
      val account = row.getAs[String]("account")
      // 使用日志输出，Worker节点日志会记录该信息
      logger.info(s"number: $number, Account: $account")
    }

    spark.stop()
  }
}

额外调试建议 #

• 快速验证DataFrame内容时，优先用df.show()，比手动遍历更高效：

  df.show()
  

• 处理数据时尽量使用Spark分布式API（如select、map等），避免将大数据拉到Driver端，保证处理效率。

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