当使用Flink时，Kafka中的json数据不清楚时，如何反序列化

Question

我想通过Flink计算Kafka中的数据，但问题是Kafka的JASON数据可能会发生变化。

像这样：

{"data":{"template":25,"name":"name"}}

或者这个：

{"data"{"type":"type1","props":"props","strem":"stream1"}

我无法提前知道这个JSON中包含了多少数据

streamExecutionEnvironment.addSource(new FlinkKafkaConsumer010<KafkaEvent>("flink", new KafkaEventSchema(),
kafkaProps))
.flatMap(new KafkaEventToRow()).returns(getReturnType());

那么，当Json数据如此时，如何定义pojo类型和mapFunation呢？

Answer 1

您必须定义一个更通用的反序列化模式，如Map

定义架构

class CustomSchema implements DeserializationSchema {
private ObjectMapper mapper = new ObjectMapper();
@Override
public Map<String,Object> deserialize(byte[] bytes) throws IOException {
Map<String,Object> t = null;
t = mapper.readValue(bytes, Map.class);
return t;
}
@Override
public boolean isEndOfStream(Object o) {
return false;
}
@Override
public TypeInformation<Map> getProducedType() {
return TypeInformation.of(new TypeHint<Map>() {
});
}
}

现在将其用作模式

streamExecutionEnvironment
.addSource(new FlinkKafkaConsumer010<KafkaEvent>("flink", new CustomSchema(),......

现在你得到了一个通用的Map，它可以包含任何数据结构