如何在Beam的PCollection中使用不可序列化的遗留Java类？

针对你遇到的问题，核心原因是部分Beam转换（如AddFields）在处理Row时，若未显式绑定Schema与逻辑类型编码器，会 fallback 到默认Java序列化逻辑。以下是可复用的解决流程及示例：

1. 实现合规的Beam逻辑类型 #

逻辑类型是Beam Schema体系中处理自定义类型的核心，需严格遵循单例模式，并完成对象与Beam可序列化类型的双向转换。

示例：结构化存储的逻辑类型（将遗留类转换为Row字段） #

假设遗留类为：

// 无法修改的遗留类，未实现Serializable
public class LegacyNonSerializableClass {
    private final String id;
    private final int value;

    public LegacyNonSerializableClass(String id, int value) {
        this.id = id;
        this.value = value;
    }

    // Getters
    public String getId() { return id; }
    public int getValue() { return value; }
}

对应的逻辑类型实现：

public class LegacyClassLogicalType extends LogicalType<LegacyNonSerializableClass, Row> {
    // 唯一标识逻辑类型的URI，需符合Beam规范
    private static final String TYPE_URI = "beam:logical_type:legacy_class:v1";
    // 定义遗留类在Row中的存储结构
    private static final Schema LEGACY_STORAGE_SCHEMA = Schema.builder()
            .addStringField("id")
            .addInt32Field("value")
            .build();

    // 单例模式（Beam要求逻辑类型必须是单例）
    private static class SingletonHolder {
        private static final LegacyClassLogicalType INSTANCE = new LegacyClassLogicalType();
    }

    public static LegacyClassLogicalType getInstance() {
        return SingletonHolder.INSTANCE;
    }

    private LegacyClassLogicalType() {
        super(TYPE_URI,
                LegacyNonSerializableClass.class,
                LEGACY_STORAGE_SCHEMA,
                Row.class);
    }

    // 将遗留类转换为Beam可序列化的Row
    @Override
    public Row toRowValue(LegacyNonSerializableClass input) {
        return Row.withSchema(LEGACY_STORAGE_SCHEMA)
                .addValue(input.getId())
                .addValue(input.getValue())
                .build();
    }

    // 从Row还原回遗留类
    @Override
    public LegacyNonSerializableClass fromRowValue(Row row) {
        return new LegacyNonSerializableClass(
                row.getString("id"),
                row.getInt32("value"));
    }
}

备选：字节数组存储的逻辑类型（适合无法结构化的遗留类） #

如果遗留类结构复杂，可直接序列化为字节数组：

public class LegacyClassBytesLogicalType extends LogicalType<LegacyNonSerializableClass, byte[]> {
    private static final String TYPE_URI = "beam:logical_type:legacy_class_bytes:v1";
    private static final ObjectMapper JSON_MAPPER = new ObjectMapper();

    private static class SingletonHolder {
        private static final LegacyClassBytesLogicalType INSTANCE = new LegacyClassBytesLogicalType();
    }

    public static LegacyClassBytesLogicalType getInstance() {
        return SingletonHolder.INSTANCE;
    }

    private LegacyClassBytesLogicalType() {
        super(TYPE_URI,
                LegacyNonSerializableClass.class,
                FieldType.BYTES,
                byte[].class);
    }

    @Override
    public byte[] toRowValue(LegacyNonSerializableClass input) {
        try {
            return JSON_MAPPER.writeValueAsBytes(input);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("Failed to serialize legacy class", e);
        }
    }

    @Override
    public LegacyNonSerializableClass fromRowValue(byte[] rowValue) {
        try {
            return JSON_MAPPER.readValue(rowValue, LegacyNonSerializableClass.class);
        } catch (IOException e) {
            throw new RuntimeException("Failed to deserialize legacy class", e);
        }
    }
}

2. 注册逻辑类型并绑定Schema #

必须将逻辑类型注册到Pipeline的编码器注册表，并显式为PCollection绑定Schema与Row编码器，避免Beam自动推断出错。

public class BeamLegacyClassDemo {
    public static void main(String[] args) {
        Pipeline pipeline = Pipeline.create();
        CoderRegistry coderRegistry = pipeline.getCoderRegistry();

        // 注册逻辑类型
        coderRegistry.registerLogicalType(LegacyClassLogicalType.getInstance());

        // 创建包含遗留类字段的Schema
        PipelineSchema pipelineSchema = Schema.builder()
                .addField("legacyObj", FieldType.logicalType(LegacyClassLogicalType.getInstance()))
                .addStringField("originalField")
                .build();

        // 构造输入数据并绑定Schema与编码器
        List<Row> inputRows = Arrays.asList(
                Row.withSchema(pipelineSchema)
                        .addValue(new LegacyNonSerializableClass("obj1", 100))
                        .addValue("test1")
                        .build(),
                Row.withSchema(pipelineSchema)
                        .addValue(new LegacyNonSerializableClass("obj2", 200))
                        .addValue("test2")
                        .build()
        );

        PCollection<Row> input = pipeline.apply(Create.of(inputRows))
                .setRowSchema(pipelineSchema)
                .setCoder(RowCoder.of(pipelineSchema)); // 显式设置编码器
    }
}

3. Schema-aware的转换操作 #

对于AddFields这类Schema修改操作，必须显式指定新Schema，确保Beam始终使用逻辑类型编码器处理Row，而非默认序列化。

// 创建包含新字段的Schema
Schema newSchema = pipelineSchema.toBuilder()
        .addInt32Field("newField")
        .build();

// 执行AddFields并绑定新Schema
PCollection<Row> withNewField = input.apply(AddFields.create(newSchema,
        Field.of("newField", FieldType.INT32, 999) // 默认值999
));

// 验证转换结果
withNewField.apply("Print Updated Rows", ParDo.of(new DoFn<Row, Void>() {
    @ProcessElement
    public void process(@Element Row row, OutputReceiver<Void> out) {
        LegacyNonSerializableClass obj = row.getValue("legacyObj");
        System.out.printf("Updated: Legacy ID=%s, New Field=%d%n",
                obj.getId(), row.getInt32("newField"));
    }
}));

pipeline.run().waitUntilFinish();

关键注意事项 #

• 逻辑类型必须是单例：Beam会缓存逻辑类型实例，非单例会导致类型识别失败。
• 显式绑定编码器：所有涉及Row的PCollection都要调用setCoder(RowCoder.of(schema))，禁用自动推断。
• 避免直接暴露遗留对象：不要将遗留类作为PCollection的元素类型，始终包裹在Row的逻辑类型字段中。

内容的提问来源于stack exchange，提问作者J. Bloom