ParDo反射使用Protobuf类型触发NotSerializableException问题咨询

安全阿岳

2026-8-29

根本原因

Apache Beam 提交作业时会将所有DoFn实例序列化为字节码，分发到分布式Worker节点后再反序列化执行，因此DoFn的所有实例成员字段必须实现java.io.Serializable接口，否则序列化阶段就会抛出异常。
报错的核心原因是你在第二个DoFn中将com.google.protobuf.Descriptors.Descriptor类型的对象作为普通实例成员存储：Descriptors.Descriptor类本身没有实现序列化接口，无法被Java序列化，直接触发栈中打印的NotSerializableException。
第一个DoFn能正常运行，本质是你没有在DoFn构造阶段就将Descriptor赋值给普通实例字段（通常是将Descriptor初始化逻辑放在了@Setup等Worker端执行的生命周期方法中，或者根本没有持久化存储Descriptor实例），因此序列化时不存在不可序列化的字段，自然不会报错。

修复方案

核心思路是将不可序列化的Descriptor对象从DoFn的序列化流程中剥离，不要在构造DoFn实例时就初始化Descriptor字段：
1. 将Descriptor对应的成员变量添加transient关键字修饰，标记该字段不需要参与Java序列化流程
2. 仅将可序列化的Protobuf类标识（类全限定名字符串、Protobuf对应的Class对象，二者均支持序列化）作为DoFn的构造参数传入，存为普通实例成员
3. 将Descriptor的初始化逻辑移动到DoFn的@Setup生命周期方法中：该方法会在Worker节点完成DoFn反序列化、正式开始处理数据前执行，完全绕开作业提交阶段的DoFn序列化步骤
参考实现代码：

public class ProcessExtractedDataFn extends DoFn<byte[], Message> {
    // 仅存储可序列化的Protobuf类名
    private final String targetProtoClassName;
    // 标记为transient，不参与序列化
    private transient Descriptor targetProtoDescriptor;

    public ProcessExtractedDataFn(String targetProtoClassName) {
        this.targetProtoClassName = targetProtoClassName;
    }

    @Setup
    public void initDescriptor() throws ClassNotFoundException {
        // Worker端初始化时才加载类、获取Descriptor
        Class<?> protoClazz = Class.forName(targetProtoClassName);
        this.targetProtoDescriptor = MessageReflector.getDescriptor(protoClazz);
    }

    @ProcessElement
    public void process(@Element byte[] inputBytes, OutputReceiver<Message> output) {
        // 直接使用targetProtoDescriptor完成Protobuf反序列化即可
        // ... 原有业务逻辑
    }
}

额外适配注意：由于你将DoFn的输出类型声明为顶层的com.google.protobuf.Message，Beam无法自动推断输出PCollection的Coder，需要在应用该DoFn后显式为输出PCollection设置对应Protobuf类型的ProtoCoder，避免后续出现Coder缺失的报错。

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