Spark Encoders.bean处理泛型类失效问题求助

我有一个泛型Java类：

public class Person<T> {
    private String name;
    private List<T> attributes;
}

public class AttributeOne {
    // some fields
}

public class AttributeTwo {
    // some fields
}

我希望将Spark DataSet转换为Java Person<T> 对象列表（输入DataSet的每条记录转为一个Person<T>对象）。

最初尝试向Encoder传入泛型类型"T"，代码大致如下：

val ds = inputDS.map(<将输入DataSet转为Person对象的逻辑>)(Encoders.bean(classOf[Person[T]]))

ds.write.json(...)

代码编译运行无报错，但输出仅name字段编码成功，泛型attributes字段未正常编码，输出数据如下：

{"name": "name-1", "attributes": [{}, {}, {}]}
{"name": "name-2", "attributes": [{}, {}]}
...

Encoder能识别attributes列表的元素数量，但每个元素都是空对象{}。

意识到Encoder需要编译时确定确切类型后，传入了具体类型：

val ds = inputDS.map(<将输入DataSet转为Person<AttributeOne>对象的逻辑>)(Encoders.bean(classOf[Person[AttributeOne]]))

ds.write.json(...)

结果仍相同——name字段正常，attributes字段还是空对象（元素数量能识别）。

1. 用ScalaReflection构建带泛型信息的Encoder #

Java的类型擦除会导致Encoders.bean()无法获取泛型参数的具体类型，需要通过ScalaReflection保留泛型信息来构建正确的Encoder：

import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.Encoder

// 构建包含具体泛型的Encoder
val personEncoder: Encoder[Person[AttributeOne]] = ScalaReflection
  .encoderFor[Person[AttributeOne]]
  .asInstanceOf[Encoder[Person[AttributeOne]]]

val ds = inputDS.map(<转换逻辑>)(personEncoder)
ds.write.json(...)

2. 确保泛型元素类符合Java Bean规范 #

检查AttributeOne/AttributeTwo类是否满足：

• 存在无参构造函数
• 所有需要序列化的字段都有对应的getXxx()和setXxx()方法
• 字段类型为Spark支持的序列化类型（基本类型、String、其他合规Bean类）

如果缺少getter/setter，Spark的Bean Encoder无法读取字段值，自然会输出空对象。

3. 显式指定列表元素的Encoder #

在转换逻辑中，针对attributes列表的元素类型单独指定Encoder，确保Spark能正确序列化内部对象：

import org.apache.spark.sql.Encoders

// 获取元素类型的Encoder
val attrOneEncoder = Encoders.bean(classOf[AttributeOne])
// 构建Person<AttributeOne>的Encoder
val personEncoder = ScalaReflection.encoderFor[Person[AttributeOne]].asInstanceOf[Encoder[Person[AttributeOne]]]

val ds = inputDS.map { row =>
  val name = row.getAs[String]("name")
  // 手动转换列表元素并确保类型正确
  val attrs = row.getAs[Seq[Row]]("attributes").map { attrRow =>
    val attr = new AttributeOne()
    attr.setField1(attrRow.getAs[String]("field1"))
    attr.setField2(attrRow.getAs[Int]("field2"))
    attr
  }
  val person = new Person[AttributeOne]()
  person.setName(name)
  person.setAttributes(attrs)
  person
}(personEncoder)

4. 改用Scala Case Class（可行的话） #

Scala的Case Class能更好地被Spark的Encoder识别，泛型信息保留更完善，无需手动构建Encoder：

// 定义Scala版的实体类
case class AttributeOne(field1: String, field2: Int)
case class Person[T](name: String, attributes: List[T])

// 直接转换，Spark自动生成正确的Encoder
val ds = inputDS.map { row =>
  Person(
    name = row.getAs[String]("name"),
    attributes = row.getAs[Seq[AttributeOne]]("attributes").toList
  )
}
ds.write.json(...)

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