Flink中KinesisStreamsSink的SerializationSchema open方法未调用求助

方法一：改用RichSerializationSchema #

Flink的SerializationSchema本身没有open生命周期方法，你需要让自定义Schema继承**RichSerializationSchema**——它继承了SerializationSchema和RichFunction，会被Flink运行时自动调用open方法。

步骤： #

1. 修改CustomizedSchema的定义，继承RichSerializationSchema<String>：

public class CustomizedSchema extends RichSerializationSchema<String> {
    // 用transient标记非序列化成员变量，避免序列化报错
    private transient NonSerializableClass nonSerializableObj;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 在这里完成非序列化类的初始化逻辑
        nonSerializableObj = new NonSerializableClass();
    }

    @Override
    public byte[] serialize(String element) {
        // 使用初始化后的实例完成序列化
        return nonSerializableObj.convertToBytes(element);
    }
}

2. 原有的KinesisStreamsSink构建代码无需修改，直接传入new CustomizedSchema()即可——因为RichSerializationSchema是SerializationSchema的子类。

方法二：懒加载初始化（备选） #

如果无法改用RichSerializationSchema，可以在serialize方法中实现线程安全的懒加载初始化，确保第一次序列化时完成非序列化类的初始化：

public class CustomizedSchema implements SerializationSchema<String> {
    private transient NonSerializableClass nonSerializableObj;
    private final Object initLock = new Object();

    @Override
    public byte[] serialize(String element) {
        // 双重检查锁实现线程安全的懒加载
        if (nonSerializableObj == null) {
            synchronized (initLock) {
                if (nonSerializableObj == null) {
                    nonSerializableObj = new NonSerializableClass();
                }
            }
        }
        return nonSerializableObj.convertToBytes(element);
    }
}

注意事项 #

• 确保使用的Flink版本≥1.13，早期版本的Kinesis Sink可能未完全支持RichSerializationSchema的生命周期方法调用。
• 非序列化类的成员变量必须用transient修饰，避免Flink序列化算子时抛出异常。

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