Configurable Parser

[Theo]

Hi there,

In my usecase, I read data from Kafka where in each kafka partition, I have ascending timestamps.

Currently, I parse the data from Kafka with a custom deserialization schema so that after parsing, the FlinkKafkaConsumerBase can extract the eventtime ascending timestamps and create proper watermarks via the WatermarkMultiplexer (i.e. take the minimum watermark over all non-idling assigned partitions in that task).

Now, we have a strange modification that the parser can change at runtime and only via the new parser, I can extract the timestamp field of the received byte[]. The parser change is told to me via another kafka topic.

I immediately thought about: That's a perfect usecase for broadcast streams: I connect the parser config stream and buffer the events in the connect function if the old parser is not able to parse the events up until a new parser arrives. (Doesn't sound too good from architecture all in all, but that's how it is).

My problem is the following: If I want to use broadcast stream, I must outsource my parser to a new pipeline step and don't parse within the KafkaDeserializationSchema any longer. This also means that Flink/Kafka can't produce the watermarks and I need to emulate the nice per partition ascending watermark assigner with the downstream multiplexer myself. Am I correct? Can I "easily" plugin to my stream (after broadcast parsing) this timestamp assigner with multiplexer logic? Could it also detect idle partitions like the KafkaConsumer? Or which way would you go? The only alternative I see is to greatly incrase the complexity of my KafkaDeserializationSchema to also read another kafka topic in background and as well buffer elements internally.. Sounds not very "flinkish".

Best regards

Theo

[Timo Walther]

Hi Theo,

this is indeed a difficult use case. The KafkaDeserializationSchema is
actually meant mostly for deserialization and should not contain more
complex logic such as joining with a different topic. You would make
KafkaDeserializationSchema stateful.

But in your usecase, I see no better alternative than making
KafkaDeserializationSchema more complex if per-partition watermarking
should be in place. Are the parser changes happen frequently? And how do
you deal with a failure case when the Flink job restarts and all state
in the deserialization schema is lost? You might need to extend the
FlinkKafkaConsumer to add another parser state for persistence.

Regards,
Timo


On 21.10.20 17:25, Theo Diefenthal wrote:

> Hi there,
>
> In my usecase, I read data from Kafka where in each kafka partition, I
> have ascending timestamps.
> Currently, I parse the data from Kafka with a custom deserialization
> schema so that after parsing, the FlinkKafkaConsumerBase can extract the
> eventtime ascending timestamps and create proper watermarks via the
> WatermarkMultiplexer (i.e. take the minimum watermark over all
> non-idling assigned partitions in that task).
>
> Now, we have a strange modification that the parser can change at
> runtime and only via the new parser, I can extract the timestamp field
> of the received byte[]. The parser change is told to me via another
> kafka topic.
> I immediately thought about: That's a perfect usecase for broadcast
> streams: I connect the parser config stream and buffer the events in the
> connect function if the old parser is not able to parse the events up
> until a new parser arrives. (Doesn't sound too good from architecture
> all in all, but that's how it is).
>
> My problem is the following: If I want to use broadcast stream, I must
> outsource my parser to a new pipeline step and don't parse within the
> KafkaDeserializationSchema any longer. This also means that Flink/Kafka
> can't produce the watermarks and I need to emulate the nice per
> partition ascending watermark assigner with the downstream multiplexer
> myself. Am I correct? Can I "easily" plugin to my stream (after
> broadcast parsing) this timestamp assigner with multiplexer logic? Could
> it also detect idle partitions like the KafkaConsumer? Or which way
> would you go? The only alternative I see is to greatly incrase the
> complexity of my KafkaDeserializationSchema to also read another kafka
> topic in background and as well buffer elements internally.. Sounds not
> very "flinkish".
>
> Best regards
> Theo