英文:
How to provide TypeInformation for a GenericRowData type Objects in Flink
问题
Here's the translated code part you requested:
public class KafkaDeserializer implements DeserializationSchema<RowData> {
private transient ObjectMapper mapper;
@Override
public void open(InitializationContext context) {
mapper = JacksonMapperFactory.createObjectMapper();
}
@Override
public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
DeserializationSchema.super.deserialize(message, out);
}
public RowData deserialize(byte[] message) throws IOException {
JsonNode messagePayload = mapper.readTree(message);
GenericRowData genericRowData = GenericRowData.of(
StringData.fromString(messagePayload.get("x")),
StringData.fromString(messagePayload.get("Y")),
StringData.fromString(messagePayload.get("Z"))
// ...........
);
}
public TypeInformation<RowData> getProducedType() {
return TypeInformation.of(new TypeHint<RowData>() {})
}
}
If you have any more specific questions or need further assistance with this code, please feel free to ask.
英文:
I am using a deserializer to parse a Kafka Stream (of JSON Strings) and I'm then using the GeneericRowData class to convert the Object node to type RowData instance, which is supported by hudi to write directly from DataStream. I'm expected to provide the TypeInformation, so that Flink doesn't fall back to generic serialization but I'm unable to find a way, as of how to provide a TypeInformation of that matches with the GenericRowData type. I tried using TypeHint but that doesn't work.
Here's a sample code for what I'm trying to achieve.
public class KafkaDeserializer implements DeserializationSchema<RowData> {
private transient ObjectMapper mapper;
@Override
public void open(InitializationContext context) {
mapper = JacksonMapperFactory.createObjectMapper();
}
@Override
public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
DeserializationSchema.super.deserialize(message, out);
}
public RowData deserialize(byte[] message) throws IOException {
JsonNode messagePayload = mapper.readTree(message);
GenericRowData genericRowData = GenericRowData.of(
StringData.fromString(messagePayload.get("x")),
StringData.fromString(messagePayload.get("Y")),
StringData.fromString(messagePayload.get("Z"))
...........
);
}
public TypeInformation<RowData> getProducedType() {
return TypeInformation.of(new TypeHint<>() {})
}
}
Need some guidance on how can I implement the above function getProducedType().
答案1
得分: 1
只提供TypeInformation并不足以让Flink选择一个“好”的序列化器。TypeInformation必须返回一个除了Kryo之外的序列化器，而我相当确定这不会顺利运行。
你可以实现自己的GenericRowDataTypeInfo类，该类扩展了TypeInformation<GenericRowData>，并且从createSerializer()方法返回Flink的RowDataSerializer。
英文:
Just providing TypeInformation isn't enough to for Flink to pick a "good" serializer. The TypeInformation has to return a serializer other than Kryo for GenericRowData, and I'm pretty sure that won't just work out of the box.
You could implement your own GenericRowDataTypeInfo class that extends TypeInformation<GenericRowData>, and returns Flink's RowDataSerializer from the createSerializer() method.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论