Spark 3.5.0 Scala中Option[LocalDate]序列/映射编码异常求助

问题原因 #

Spark 3.5.0的默认Scala编码器对嵌套在Seq中的Option[LocalDate]处理存在兼容性问题：

• 对于Double、String等基础类型，编码器能自动解包Option，识别内部的实际类型；
• 但LocalDate属于Java 8日期类型，编码器在解析Seq[Option[LocalDate]]时，会错误地将每个元素识别为scala.Some对象，而非解包后的LocalDate或null，导致抛出scala.Some is not a valid external type for schema of date异常。

解决方案 #

方案1：转换为Spark原生支持的java.sql.Date #

这是最简便的方案，直接将LocalDate转换为Spark原生兼容的java.sql.Date，编码器能正常处理嵌套的Option和Seq结构：

case class SubClass[T](i: Option[T])

case class TestClass[T](i: Option[T] = None, st: Option[SubClass[T]], sq: Option[Seq[Option[T]]] = None)

it should "handle sequence of date with java.sql.Date" in {
  val sparkSession = spark
  import sparkSession.implicits._
  // 将LocalDate转为java.sql.Date
  val aDate = java.sql.Date.valueOf(LocalDate.parse("2024-04-29"))
  
  val ds = Seq(TestClass[java.sql.Date](
    Some(aDate),
    Some(SubClass[java.sql.Date](Some(aDate))),
    Some(Seq(Some(aDate)))
  )).toDS    
  
  ds.show
}

方案2：自定义Seq[Option[LocalDate]]编码器 #

如果必须保留LocalDate类型，可以编写自定义编码器，手动处理Option的解包和日期类型的转换逻辑：

import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.types._
import java.time.LocalDate

// 为Seq[Option[LocalDate]]编写自定义编码器
implicit val seqOptionLocalDateEncoder: ExpressionEncoder[Seq[Option[LocalDate]]] = {
  val schema = ArrayType(DateType, containsNull = true)
  ExpressionEncoder[Seq[Option[LocalDate]]](
    schema,
    flat = false,
    nullSafe = true,
    // 序列化：将Seq[Option[LocalDate]]转为Spark兼容的日期数组
    (row, value) => {
      val dateArray = value.map {
        case Some(localDate) => java.sql.Date.valueOf(localDate)
        case None => null
      }.toArray
      row.update(0, dateArray)
    },
    // 反序列化：将Spark日期数组转回Seq[Option[LocalDate]]
    (row) => {
      val arr = row.getArray(0)
      (0 until arr.numElements()).map { idx =>
        Option(arr.getDate(idx)).map(_.toLocalDate)
      }.toSeq
    }
  )
}

// 测试代码
it should "handle sequence of date with custom encoder" in {
  val sparkSession = spark
  import sparkSession.implicits._
  val aDate = LocalDate.parse("2024-04-29")
  
  val ds = Seq(TestClass[LocalDate](
    Some(aDate),
    Some(SubClass[LocalDate](Some(aDate))),
    Some(Seq(Some(aDate), None)) // 测试包含None的场景
  )).toDS    
  
  ds.show
}

方案3：手动转换结构后转DataFrame #

先将嵌套的Option[LocalDate]结构转换为Spark兼容的扁平格式，再构建DataFrame后转为Dataset：

it should "handle sequence of date via DataFrame conversion" in {
  val sparkSession = spark
  import sparkSession.implicits._
  val aDate = LocalDate.parse("2024-04-29")
  
  // 先转换为Spark兼容的类型（java.sql.Date）
  val rawData = Seq(
    (
      Some(java.sql.Date.valueOf(aDate)),
      Some(SubClass[java.sql.Date](Some(java.sql.Date.valueOf(aDate)))),
      Some(Seq(Some(java.sql.Date.valueOf(aDate)), None))
    )
  )
  
  // 构建DataFrame后转为Dataset
  val ds = spark.createDataFrame(rawData)
    .toDF("i", "st", "sq")
    .as[TestClass[LocalDate]]
  
  ds.show
}

验证结果 #

上述三种方案均能解决Seq[Option[LocalDate]]的编码异常，其中方案1（转换为java.sql.Date）实现成本最低，适合大多数场景；如果业务必须保留LocalDate类型，方案2的自定义编码器是更合适的选择。

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