Scala/Spark读取CSV遇NullPointerException，求Null转0的有效方案

我来帮你解决这个问题——Spark处理CSV时遇到Null值引发NullPointerException的情况很常见，你之前用if else没生效大概率是因为直接操作Row时没正确处理Spark的Null类型（注意：Spark的Null和Scala原生的null不是一回事）。下面给你几个高效可行的方案，从源头到后续处理全覆盖：

方案一：读取阶段定义Schema + 批量替换Null（推荐） #

这种方法在读取CSV时就明确数据结构，再用Spark内置的API批量替换指定列的Null为0，完全避免手动处理Row的风险：

import org.apache.spark.sql.types._
import org.apache.spark.sql.SparkSession

// 初始化SparkSession（如果还没初始化的话）
val spark = SparkSession.builder()
  .appName("HandleCSVNulls")
  .master("local[*]") // 本地测试用，生产环境去掉
  .getOrCreate()

import spark.implicits._

// 定义你的CSV Schema，根据实际列调整类型和可空性
val csvSchema = StructType(Seq(
  StructField("col1", StringType, nullable = false), // 假设第一列是不可空的字符串
  StructField("col2", IntegerType, nullable = true), // 第二列是可能为Null的数值列
  StructField("col3", DoubleType, nullable = true)   // 第三列也是可能为Null的数值列
))

// 读取CSV文件，指定Schema和表头选项（如果你的CSV有表头的话）
val rawDF = spark.read
  .option("header", "true")
  .option("nullValue", "") // 如果CSV里用空字符串表示Null，加上这个选项
  .schema(csvSchema)
  .csv("/path/to/your/csv/file.csv")

// 将指定列的Null值替换为0，Map的key是列名，value是替换值
val processedDF = rawDF.na.fill(Map(
  "col2" -> 0,
  "col3" -> 0.0 // 注意数值类型要匹配，Double列用0.0
))

// 输出结果，验证Null是否被替换
processedDF.show()

方案二：用when/otherwise精准替换单列 #

如果只需要对特定列做Null替换，或者要添加额外的判断逻辑，用when函数更灵活：

import org.apache.spark.sql.functions._

// 只把col2的Null替换为0，其他列保持原样
val processedDF = rawDF.withColumn(
  "col2",
  when(col("col2").isNull, 0).otherwise(col("col2"))
)

// 如果需要同时处理多列，可以链式调用withColumn
val processedDFMulti = processedDF.withColumn(
  "col3",
  when(col("col3").isNull, 0.0).otherwise(col("col3"))
)

为什么你之前的if else没生效？ #

如果是在RDD层面手动处理Row，直接调用row.getInt(index)会在Null值时抛出NPE，因为Spark的Null在Row中是用None表示的，不是Scala的null。如果一定要用RDD处理，正确的写法是：

// RDD层面的正确处理方式（不推荐，除非有特殊业务需求）
val processedRDD = rawDF.rdd.map { row =>
  val col1 = row.getString(0)
  // 用Option包裹，避免NPE，Null时返回0
  val col2 = row.getAs[Option[Int]]("col2").getOrElse(0)
  val col3 = row.getAs[Option[Double]]("col3").getOrElse(0.0)
  (col1, col2, col3)
}

// 转成DataFrame输出
processedRDD.toDF("col1", "col2", "col3").show()

总结 #

优先用DataFrame的na.fill或when/otherwise API，它们是Spark优化过的内置方法，代码更简洁，性能也更好，还能避免手动处理Row时的NPE问题。

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