Spark 2.4.1自定义String基UDT从CSV读取失败问题排查

你遇到的这个ClassCastException确实是Spark 2.x版本中处理UDT与CSV数据源交互时的典型问题，和你推测的原因完全一致——UnivocityParser在处理UDT类型时生成的转换器逻辑不符合预期，导致后续类型转换失败。

问题根源拆解 #

当你定义了基于StringType的自定义UDTMyType并关联到MyValue类后，Spark使用Univocity解析CSV时，没有正确生成适配UDT的转换器：

• 正常情况下，针对普通字符串类型，解析器会返回String => Any的函数，直接将CSV中的字符串转换为对应类型；
• 但对于UDT，它错误地生成了嵌套的String => (String => Any)函数，这就导致后续在调用getUTF8String时，试图把一个函数对象强转为UTF8String，最终抛出ClassCastException。

可行解决方案 #

1. 自定义CSV转换器绕过默认逻辑

你可以手动为MyValue类型注册一个自定义CSV转换器，替换Univocity的默认处理：

import org.apache.spark.sql.execution.datasources.csv.CSVConverter
import org.apache.spark.sql.types.DataType
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField

// 实现自定义转换器
class MyValueCSVConverter extends CSVConverter {
  override def convert(value: String): Any = {
    // 根据你的MyValue构造逻辑调整，这里假设MyValue接受一个String参数
    MyValue(value)
  }
}

// 注册转换器到Spark
val customConverters = Map[DataType, CSVConverter](
  new MyType() -> new MyValueCSVConverter()
)

// 读取CSV时指定自定义转换器
val df = spark.read
  .option("header", "true")
  .schema(StructType(Array(StructField("target_col", new MyType()))))
  .format("csv")
  .option("csv.converter", customConverters)
  .load("your/csv/path")

2. 临时 workaround：先读成String再转换

如果不想自定义转换器，可以先把字段读取为普通StringType，再通过UDF转换为MyValue类型：

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

// 定义转换UDF
val strToMyValue = udf((s: String) => MyValue(s))

// 读取CSV后转换字段类型
val df = spark.read
  .option("header", "true")
  .csv("your/csv/path")
  .withColumn("target_col", strToMyValue(col("target_col")))

3. 升级Spark版本彻底解决

这个问题在Spark 3.x版本中已经被官方修复，因为Spark 3.x对UDT的整体处理逻辑进行了重构，UnivocityParser的转换器生成逻辑也做了针对性优化。如果项目条件允许，升级到Spark 3.0+版本可以一劳永逸解决这个问题。

额外注意点 #

确保你的MyType实现了完整的UDT要求，尤其是serialize和deserialize方法要正确处理MyValue与底层字符串/UTF8String的转换：

class MyType extends UserDefinedType[MyValue] {
  override def sqlType: DataType = StringType

  override def serialize(obj: MyValue): Any = obj.toString // 根据实际业务调整

  override def deserialize(datum: Any): MyValue = datum match {
    case s: String => MyValue(s)
    case utf8: UTF8String => MyValue(utf8.toString)
    case _ => throw new IllegalArgumentException(s"无法将 $datum 反序列化为 MyValue")
  }

  override def userClass: Class[MyValue] = classOf[MyValue]
}

内容的提问来源于stack exchange，提问作者Kal-ko