Flink Kafka - 自定义类数据始终为空答案

【问题标题】：Flink Kafka - Custom Class Data is always nullFlink Kafka - 自定义类数据始终为空
【发布时间】：2018-12-10 21:14:55
【问题描述】：

自定义类

人物

class Person
{
  private Integer id;
  private String name; 
 //getters and setters
}

Kafka Flink 连接器

TypeInformation<Person> info = TypeInformation.of(Person.class);
TypeInformationSerializationSchema schema = new TypeInformationSerializationSchema(info, new ExecutionConfig());
DataStream<Person> input = env.addSource( new FlinkKafkaConsumer08<>("persons", schema , getKafkaProperties()));

现在如果我发送下面的 json

{ "id" : 1, "name": Synd }

通过Kafka Console Producer，flink代码抛出空指针异常 但是，如果我使用 SimpleStringSchema 而不是之前定义的 CustomSchema，则会打印流。

上面的设置有什么问题

【问题讨论】：

标签： json serialization apache-kafka deserialization apache-flink

【解决方案1】：

TypeInformationSerializationSchema 是一个反序列化模式，它使用 Flink 的序列化堆栈，因此也使用它的序列化器。因此，当使用这个SerializationSchema时，Flink 期望数据已经被 Flink 的序列化器序列化为Person 类型。

鉴于Person 类的摘录，Flink 很可能会使用它的PojoTypeSerializer。此序列化程序无法理解提供 JSON 输入数据。

如果您想使用 JSON 作为输入格式，那么您必须定义自己的 DeserializationSchema，它可以将 JSON 解析为 Person。

【讨论】：

成功了！更好的文档会有所帮助

【解决方案2】：

为有相同问题的人解答

自定义序列化程序

class PersonSchema implements DeserializationSchema<Person>{

    private ObjectMapper mapper = new ObjectMapper(); //com.fasterxml.jackson.databind.ObjectMapper;

    @Override
    public Person deserialize(byte[] bytes) throws IOException {
        return mapper.readValue( bytes, Person.class );
    }

    @Override
    public boolean isEndOfStream(Person person) {
        return false;
    }

    @Override
    public TypeInformation<Person> getProducedType() {
        return TypeInformation.of(new TypeHint<Person>(){});
    }
}

使用架构

DataStream<Person> input = env.addSource( new FlinkKafkaConsumer08<>("persons", new PersonSchema() , getKafkaProperties()));

【讨论】：