Flink Kafka生产者：类的对象不可序列化

Question

我尝试为我的定制类实现一个方法，使用Flink-Kafka连接器生成关于Kafka的数据。类原型如下所示：

public class StreamData implements Serializable {
    private transient StreamExecutionEnvironment env;
    private DataStream<byte[]> data ;
    ...

将数据写入特定Kafka主题的方法如下：

public void writeDataIntoESB(String id) throws Exception {

        FlinkKafkaProducer011<byte[]> producer = new FlinkKafkaProducer011<byte[]>(
                "localhost:9092",
                id,
                new KeyedSerializationSchema<byte[]>() {
                    @Override
                    public byte[] serializeKey(byte[] bytes) {
                        return bytes;
                    }

                    @Override
                    public byte[] serializeValue(byte[] bytes) {
                        return bytes;
                    }

                    @Override
                    public String getTargetTopic(byte[] bytes) {
                        return null;
                    }
                });               
        data.addSink(producer);
    }

我有另一种方法可以从Kafka主题获取对象的data字段中的数据，效果很好。现在尝试从Kafka主题获取数据并将其写入另一个Kafka主题时，我遇到了错误：

org.apache.flink.api.common.InvalidProgramException: Object StreamData$2@1593948d is not serializable

主要代码：

StreamData temp = new StreamData();
temp = temp.getDataFromESB("data", 0);
temp.writeDataIntoESB("flink_test");

Java似乎试图序列化对象，而不仅仅是字段数据！使用Flink Kafka连接器向Kafka生成数据的代码已经过测试，并且可以正常使用（我的意思是不使用类，而是在main中编写所有代码）

如何消除错误？




Answer 1

使数据属性静态化，解决了问题。有人能详细说明这一点吗？这是一个好的解决方案吗？

private static DataStream<byte[]> data ;

Answer 2

您也可以像这样在Flink中进行序列化

dataStream.addSink(new FlinkKafkaProducer<KafkaObject>(ProducerTopic, new 
                                             CustomSerializerSchema(),properties));

  public class CustomSerializerSchema implements SerializationSchema<MyUser> {

    private static final long serialVersionUID = 1L;

    @Override
    public byte[] serialize(MyUser element) {
        byte[] b = null;
         try {
             b= new ObjectMapper().writeValueAsBytes(element);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return b; 
    }
}

Answer 3

我认为问题的原因是您的代码正在执行以下操作：

new KeyedSerializationSchema<byte[]>() {...}

这段代码所做的是创建一个匿名的KeyedSeriazationSchema子类，作为定义类（StreamData）的内部类。每个内部类都持有对外部类实例的隐式引用，因此使用默认Java序列化规则序列化它也会传递地尝试序列化外部对象（StreamData）。解决此问题的最好方法是将您的KeyedSeriazationSchema子类声明为：

• 新的顶级课程或

我认为最后一种方法是这样的：

public class StreamData {
    static KeyedSerializationSchema<byte[]> schema = new KeyedSerializationSchema<byte[]>() {
        ...
    };
    ...
    public void writeDataIntoESB(String id) throws Exception {

        FlinkKafkaProducer011<byte[]> producer = new FlinkKafkaProducer011<byte[]>("localhost:9092", id, schema);               
        data.addSink(producer);
    }
}