Hi all, I’m reaching out today for some suggestions (and hopefully a solution) for a Flink job that I’m working on. The job itself reads JSON strings from a Kafka topic and reads those into JSONObjects (currently via Gson), which are then operated against, before ultimately being written out to Kafka again. The problem here is that the shape of the data can vary wildly and dynamically. Some records may have properties unique to only that record, which makes defining a POJO difficult. In addition to this, the JSONObjects fall by to Kryo serialization which is leading to atrocious throughput. I basically need to read in JSON strings, enrich properties on these objects, and ultimately write them to various sinks. Is there some type of JSON-based class or library or an approach I could use to accomplish this in an efficient manner? Or if possibly a way to partially write a POJO that would allow me to interact with sections/properties of the JSON while retaining other properties that might be dynamically present or unique to the message? Any advice or suggestions would be welcome! I’ll also be happy to provide any additional context if it would help!

The problem here is that the shape of the data can vary wildly and dynamically.

I’d still try to use schematized events here, e.g. with Avro (unless you’re talking about millions of potential schemas). So on the producer side, create a schema for each type of event, register it in the schema registry, etc. On the consumer side (Flink) you can deserialize it into Avro’s GenericRecords and then decide how to process them depending on the present fields, etc.

So one of the challenges here is that this specific application doesn’t have any control over the topic that is being written to. It has a large (unknown number) of producers that each have their own shape and a few guaranteed similar fields that just come into the topic as JSON strings. Do you know if it’s possible to take the JSON strings and leverage Avro’s GenericRecords to handle this? Essentially the pipeline is strings in, strings out, all JSON.

this specific application doesn’t have any control over the topic that is being written to. It has a large (unknown number) of producers that each have their own shape and a few guaranteed similar fields that just come into the topic as JSON strings.

I don’t think that writing to a certain topic is relevant here. But I think cleaning up the producers and adding more structure there will benefit you more long-term.

Oh I agree, I’m just saying in this situation the producers themselves are external to this system and not really able to be adjusted in this case. It’s a bit of having to take whatever is provided that has some rough shape but some messages may contain fields that others don’t and those need to be retained (although almost all messages have a subset of the same fields).

It seems like your idea is to generically parse every type of message from Kafka stream and followup processing will take care of data normalization. This an unusual scenario, but I am guessing your process function has some kind of validation requirements on the input. I would recommend to create a message filter for serialized String data by converting the validations into regex. That would leave you with only a subset of input to deserialize. In this subset you can determine patterns and create POJO for each pattern. The detection of POJO could be done through a factory interface.