Spark Scala调用Main类Dataset函数时抛出NullPointerException

我在Test测试类里调用Main主应用类中定义的nr_drives_by_month函数，该函数返回一个由Main类内部定义的driveAggMonthly类构成的Dataset，目的是把返回数据和Test类里创建的预期Dataset做对比，但执行时抛出空指针异常：

java.lang.NullPointerException was thrown.
java.lang.NullPointerException
    at assignment.Main$.nr_drives_by_month(Main.scala:96)

简化后的代码 #

Main类 #

package assignment
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.sql.functions.to_date
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
import java.sql.Date


object Main extends App {

  Logger.getLogger("org").setLevel(Level.ERROR)

  val drive_data_location = "data/input/driveData.csv"
  val person_data_location = "data/input/person.csv"

  val spark = SparkSession
    .builder
    .master("local[1]")
    .appName("driver_analytics")
    .getOrCreate()

  spark.conf.set("spark.sql.shuffle.partitions", 1)

  import spark.implicits._

  // define schema for a person
  case class Person(personId: Int,
                             firstName: String,
                             lastName: String)

  // define schema for a drive
  case class Drive(personId: Int,
                          driveId: Int,
                          date: String)

  case class driveAggMonthly(Month: Int,
                              numberOfDrives: BigInt)


  val drives = spark.read
    .option("header", "true")
    .option("charset", "UTF8")
    .option("delimiter", ",")
    .option("dateformat", "yyyy-MM-dd")
    .option("inferSchema", "true")
    .csv(drive_data_location)
    .as[Drive]
    .cache()

  val persons = spark.read
    .option("header", "true")
    .option("charset", "UTF8")
    .option("delimiter", ",")
    .option("inferSchema", "true")
    .csv(person_data_location)
    .as[Person]
    .cache()


  def nr_drives_by_month(input_ds: Dataset[Drive]): Dataset[driveAggMonthly] = {
    input_ds
      .groupBy(month(to_date(col("date"))).as("Month"))
      .agg(count(col("driveId")).as("numberOfDrives"))
      .as[driveAggMonthly]
  }

}

Test类 #

package assignment

import assignment.Main._
import org.apache.spark.sql.{Dataset, SparkSession}
import org.scalatest.funsuite.AnyFunSuite

class MainTest extends AnyFunSuite {

  val spark = SparkSession
    .builder
    .master("local[1]")
    .appName("driver_analytics")
    .getOrCreate()

  import spark.implicits._

  // Unit Tests:
  test("monthlyAgg") {

    val source_ds = Seq(
      Drive(1, 5, "2023-01-01"),
      Drive(2, 6, "2023-02-01"),
      Drive(3, 7, "2023-02-05"),
      Drive(4, 8, "2024-03-01")
    ).toDS()

    val result_ds_t1: Dataset[driveAggMonthly] = nr_drives_by_month(source_ds: Dataset[Drive])

    val expected_ds_t1 = Seq(
      driveAggMonthly(1, 1),
      driveAggMonthly(2, 2),
      driveAggMonthly(3, 1)
    ).toDS()
    
    assert(result_ds_t1 === expected_ds_t1)
  }
}

已知临时方案（不想采用） #

• 去掉.as[driveAggMonthly]返回DataFrame或Unit
• 将nr_drives_by_month函数移到Test类中

问题根源 #

空指针的核心原因是Main类继承了App特质：当Test类引用Main._时，会触发Main对象的完整初始化流程，而Main里的drives、persons数据集依赖本地文件路径，测试环境中这些路径不存在，导致数据集初始化失败；同时，driveAggMonthly等Case类定义在Main对象内部，依赖Main的spark.implicits._隐式转换，一旦Main初始化失败，这些隐式转换无法正常加载，最终导致as[driveAggMonthly]执行时抛出空指针。

解决方案 #

1. 重构代码：分离业务逻辑与应用入口 #

把Case类、业务函数从Main对象中抽离，避免绑定到App特质的初始化流程：

新建业务逻辑类DriverAnalytics

package assignment
import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.sql.functions._

// 将Case类移到顶层，脱离Main对象的依赖
case class Person(personId: Int, firstName: String, lastName: String)
case class Drive(personId: Int, driveId: Int, date: String)
case class driveAggMonthly(Month: Int, numberOfDrives: BigInt)

object DriverAnalytics {
  // 用隐式参数传递SparkSession，避免绑定到特定会话
  def nr_drives_by_month(input_ds: Dataset[Drive])(implicit spark: SparkSession): Dataset[driveAggMonthly] = {
    import spark.implicits._
    input_ds
      .groupBy(month(to_date(col("date"))).as("Month"))
      .agg(count(col("driveId")).as("numberOfDrives"))
      .as[driveAggMonthly]
  }
}

修改Main类为纯应用入口

package assignment
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

object Main extends App {
  Logger.getLogger("org").setLevel(Level.ERROR)

  val drive_data_location = "data/input/driveData.csv"
  val person_data_location = "data/input/person.csv"

  val spark = SparkSession
    .builder
    .master("local[1]")
    .appName("driver_analytics")
    .getOrCreate()

  spark.conf.set("spark.sql.shuffle.partitions", 1)

  import spark.implicits._
  import DriverAnalytics._

  val drives = spark.read
    .option("header", "true")
    .option("charset", "UTF8")
    .option("delimiter", ",")
    .option("dateformat", "yyyy-MM-dd")
    .option("inferSchema", "true")
    .csv(drive_data_location)
    .as[Drive]
    .cache()

  val persons = spark.read
    .option("header", "true")
    .option("charset", "UTF8")
    .option("delimiter", ",")
    .option("inferSchema", "true")
    .csv(person_data_location)
    .as[Person]
    .cache()

  // 调用业务函数
  val monthlyDrives = nr_drives_by_month(drives)
  monthlyDrives.show()
}

2. 修改Test类引用重构后的逻辑 #

package assignment

import org.apache.spark.sql.{Dataset, SparkSession}
import org.scalatest.funsuite.AnyFunSuite
import DriverAnalytics._

class MainTest extends AnyFunSuite {

  // 定义隐式SparkSession，供业务函数使用
  implicit val spark = SparkSession
    .builder
    .master("local[1]")
    .appName("driver_analytics_test")
    .getOrCreate()

  import spark.implicits._

  test("monthlyAgg") {
    val source_ds = Seq(
      Drive(1, 5, "2023-01-01"),
      Drive(2, 6, "2023-02-01"),
      Drive(3, 7, "2023-02-05"),
      Drive(4, 8, "2024-03-01")
    ).toDS()

    val result_ds_t1: Dataset[driveAggMonthly] = nr_drives_by_month(source_ds)

    val expected_ds_t1 = Seq(
      driveAggMonthly(1, 1),
      driveAggMonthly(2, 2),
      driveAggMonthly(3, 1)
    ).toDS()
    
    // 用collect()对比数组元素，避免Dataset直接===的隐式转换问题
    assert(result_ds_t1.collect().sameElements(expected_ds_t1.collect()))
  }
}

关键优势 #

• Case类移到顶层后，测试环境无需依赖Main对象的初始化，可直接访问
• 业务函数通过隐式参数传递SparkSession，适配测试和生产环境的不同会话
• 分离入口与逻辑后，测试不会触发Main中加载本地文件的代码，彻底避免路径不存在的问题

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