Apache Spark Java UDF返回Java Bean失败，咨询实现方法

可以通过Spark UDF返回Java Bean，你遇到的报错是因为注册UDF时指定了StructType作为返回类型，但UDF实际返回的是Java Bean实例，Spark无法直接完成这种类型转换。

问题根源 #

你注册UDF时用Encoders.bean(IdsBean.class).schema()（即StructType）作为返回类型，但UDF返回的是IdsBean对象，Spark内部无法自动将Java Bean实例转换成对应的Struct结构，因此抛出类型不匹配的异常。

解决方法 #

以下两种方式可以实现需求：

方式一：使用UserDefinedFunction API（推荐） #

直接用functions.udf()结合Encoders.bean()创建UDF，Spark会自动处理Java Bean与StructType的转换逻辑，代码更简洁。

修正后的代码：

// 替换原UDF注册和调用代码
UserDefinedFunction setIdUdf = functions.udf(
    (String familyName) -> {
        IdsBean ids = new IdsBean();
        ids.setId1(String.valueOf(familyName.length()));
        ids.setId2("2");
        return ids;
    },
    Encoders.bean(IdsBean.class)
);

Dataset<Row> dataset = csvds.withColumn("newCol", setIdUdf.apply(functions.col("familyName")));
dataset.show();

方式二：手动转换为Row返回 #

如果坚持使用spark.udf().register的方式，可以在UDF内部将Java Bean转换成Row，注册时指定对应的StructType：

// 定义返回Row的UDF
UDF1<String, Row> setIdUdf = new UDF1<String, Row>() {
    @Override
    public Row call(String familyName) {
        IdsBean ids = new IdsBean();
        ids.setId1(String.valueOf(familyName.length()));
        ids.setId2("2");
        // 将Bean转换为Row
        return RowFactory.create(ids.getId1(), ids.getId2());
    }
};

// 获取Bean对应的StructType
StructType idsSchema = Encoders.bean(IdsBean.class).schema();
// 注册UDF
spark.udf().register("setIdUdf", setIdUdf, idsSchema);

// 调用UDF
Dataset<Row> dataset = csvds.withColumn("newCol", functions.callUDF("setIdUdf", functions.col("familyName")));
dataset.show();

注意事项 #

• Java Bean必须包含无参构造函数和完整的getter/setter方法，这是Spark Encoder序列化Bean的必要条件（你的IdsBean已满足要求）。
• 优先选择方式一，无需手动处理Row转换，代码可读性更高。

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