Spark Java如何在DataFrame中存储自定义Java对象？

Spark DataFrame是结构化分布式数据集，不支持直接存储任意原生Java对象，所有列类型必须匹配Spark SQL的类型系统，可按以下两种方案实现需求，无需额外依赖第三方包。

前置准备：改造自定义POJO类 #

首先给Temporary类实现序列化接口、添加无参构造和对应getter/setter，这是Spark识别Java Bean、自动做类型映射的必要条件：

import java.io.Serializable;

class Temporary implements Serializable {
    private Integer id;
    private String name;
    private String another;

    // Spark反序列化Bean必须的无参构造
    public Temporary() {}

    public Temporary(Integer id, String name, String another){
        this.id = id;
        this.name = name;
        this.another = another;
    }

    // 预定义空对象实例，统一复用
    public static final Temporary EMPTY = new Temporary(null, null, null);

    // 补全所有字段的getter/setter
    public Integer getId() { return id; }
    public void setId(Integer id) { this.id = id; }
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public String getAnother() { return another; }
    public void setAnother(String another) { this.another = another; }
}

方案1：原生函数实现（性能最优，推荐） #

直接用Spark内置的struct、when/otherwise、array函数构造目标列，完全避免UDF的序列化开销，不需要额外注册函数。
你的CASE WHEN逻辑不需要拼SQL字符串传进expr，直接用Column API编写即可，从根源避免字符串转义、类识别不到的问题：

import static org.apache.spark.sql.functions.*;

// 替换成你实际的判断逻辑
Column firstCondition = col("判断字段1").equalTo("目标值");
Column secondCondition = col("判断字段2").gt(100);

// 构造第一个数组元素：条件满足则生成对应struct，否则返回null（对应空对象）
Column firstElement = when(firstCondition, struct(
        lit(1).as("id"),
        lit("john").as("name"),
        lit("tempstring").as("another")
)).otherwise(lit(null));

// 构造第二个数组元素
Column secondElement = when(secondCondition, struct(
        lit(2).as("id"),
        lit("johny").as("name"),
        lit("tempstring").as("another")
)).otherwise(lit(null));

// 组装为数组列
Dataset<Row> res = df.withColumn("Temporary", array(firstElement, secondElement));

生成的Temporary列类型为Array<Struct<id:int, name:string, another:string>>，完全符合Spark类型规范，后续如果需要把列中的struct转回Temporary Java对象，直接使用Bean编码器即可：

Encoder<Temporary> tempEncoder = Encoders.bean(Temporary.class);

方案2：UDF返回Java对象实例 #

如果你需要在逻辑中直接操作Temporary类实例，可以通过注册UDF实现，Spark会自动将返回的POJO映射为对应结构的struct类型：

import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.Encoders;

// 定义第一个元素生成逻辑：入参替换为你实际判断用到的字段类型
UDF1<String, Temporary> genFirstObj = (flagVal) -> {
    if ("expected".equals(flagVal)) {
        return new Temporary(1, "john", "tempstring");
    }
    return Temporary.EMPTY;
};

// 定义第二个元素生成逻辑
UDF1<Integer, Temporary> genSecondObj = (numVal) -> {
    if (numVal > 100) {
        return new Temporary(2, "johny", "tempstring");
    }
    return Temporary.EMPTY;
};

// 注册UDF，指定返回Schema为Temporary类对应的Struct结构
spark.udf().register("genFirstObj", genFirstObj, Encoders.bean(Temporary.class).schema());
spark.udf().register("genSecondObj", genSecondObj, Encoders.bean(Temporary.class).schema());

// 调用UDF组装数组列
Dataset<Row> res = df.withColumn("Temporary", array(
        callUDF("genFirstObj", col("判断字段1")),
        callUDF("genSecondObj", col("判断字段2"))
));

注意事项 #

• 你之前尝试在expr的SQL字符串里直接调用Temporary()类构造方法的写法不可行，是因为Spark SQL表达式引擎只能识别内置函数、已注册的UDF/UDAF，无法直接识别Java类的构造方法。
• 不要尝试直接将未经过类型映射的Java对象存入DataFrame，DataFrame底层采用专有二进制结构化格式存储，不支持存储任意Java堆对象。
• 固定长度数组中条件不满足的位置传null struct即可，和你预期的empty_object效果完全一致，后续处理时判断id字段是否为null即可识别空对象。

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