Scenario: Apache Flink, Kafka, Protocol buffers data consumer.
The data source is Kafka topics (multiple topics: topic#1, topic#3, topic#3) in protocol buffers format. The consumer is Apache Flink consumer. Each topic has a unique protobuf definitions.
List<String> topicList = Arrays.asList("topic#1,topic#2,topic#3".split(","));
inputStream = env.addSource(new FlinkKafkaConsumer[ProtobufDeserializationSchema](topicList, new ProtobufDeserializationSchema(), properties));
I am trying to develop a generic data ingestion job in Apache Flink to ingest the data from Kafka into the database.
How to implement a generic protobuf deserializer for Apache Flink? I am looking for implementation to link the Kafka topic to protobuf definition for deserialization.
The initial approach was to bring byte arrays into Flink datastream and then based on the Kafka topic name determine the protobuf definition to deserialize the messages in the map function. How can I do it in a generic fashion?
